Add the C++ event layer over libuv

src/event/loop.hpp wraps uv_loop/poll/timer/signal/async handles that
live by value inside their owner and dispatch to an owner member
function chosen at compile time. A Poll owns its socket and carries
read and write interest on one handle, closing the descriptor in
libuv's close callback. CloseGroup collects the asynchronous closes of
an owner's handles and reports completion afterwards, which is how
pooled owners implement AsyncClosable. DeadlineList drives the idle
timeouts of many owners from one timer.

test_event_cpp covers readiness, closing a peer from inside a
callback, recycling a pooled owner only after its handles closed,
repeating timers, deadline ordering and cross-thread wakeups.

.clang-tidy

@@ -1,7 +1,7 @@
 # clang-tidy configuration for shadowsocks-c project sources (src/ only).
 # Bundled sources under third_party/ and src/blake3/ are third-party code and
 # are excluded by the file regex used in CI:
-#   run-clang-tidy -p build '/(src/(core/|event/|relay/|crypto/|api/)?[^/]+\.(c|cpp)|tests/test_core_[^/]+\.cpp)$'
+#   run-clang-tidy -p build '/(src/(core/|event/|relay/|crypto/|api/)?[^/]+\.(c|cpp)|tests/test_(core_[^/]+|event_cpp)\.cpp)$'
 #
 # The header-only C++ core (src/core/*.hpp) is analysed through the
 # test_core_* translation units that include it.
@@ -39,4 +39,4 @@ Checks: >
   -clang-analyzer-security.insecureAPI.strcpy,
   -clang-analyzer-security.insecureAPI.DeprecatedOrUnsafeBufferHandling
 WarningsAsErrors: ''
-HeaderFilterRegex: '.*/src/([^/]+\.h|core/[^/]+\.hpp)$'
+HeaderFilterRegex: '.*/src/([^/]+\.h|(core|event)/[^/]+\.hpp)$'

.github/workflows/tests.yml

@@ -125,7 +125,7 @@ jobs:
         run: |
           set -o pipefail
           tidy_status=0
-          run-clang-tidy-18 -quiet -p build -extra-arg=-Wno-unknown-warning-option "$PWD/(src/(core/|event/|relay/|crypto/|api/)?[^/]+\.(c|cpp)|tests/test_core_[^/]+\.cpp)$" 2>&1 | tee tidy.log || tidy_status=$?
+          run-clang-tidy-18 -quiet -p build -extra-arg=-Wno-unknown-warning-option "$PWD/(src/(core/|event/|relay/|crypto/|api/)?[^/]+\.(c|cpp)|tests/test_(core_[^/]+|event_cpp)\.cpp)$" 2>&1 | tee tidy.log || tidy_status=$?
           grep 'warning:' tidy.log | sort -u > warnings.txt || true
           count="$(wc -l < warnings.txt)"
           echo "clang-tidy warnings: $count (max allowed: $MAX_WARNINGS)"

CLAUDE.md

@@ -76,7 +76,7 @@ Uses **uncrustify** with the config at `.uncrustify.cfg`. Key settings: 4-space
 
 - **clang-tidy**: config in `.clang-tidy`. The build exports `compile_commands.json` automatically. Run locally:
   ```bash
-  run-clang-tidy -quiet -p build '/(src/(core/|event/|relay/|crypto/|api/)?[^/]+\.(c|cpp)|tests/test_core_[^/]+\.cpp)$'
+  run-clang-tidy -quiet -p build '/(src/(core/|event/|relay/|crypto/|api/)?[^/]+\.(c|cpp)|tests/test_(core_[^/]+|event_cpp)\.cpp)$'
   ```
   On macOS with Homebrew LLVM, add `-extra-arg="-isysroot$(xcrun --show-sdk-path)"`. CI (`clang-tidy` job in `tests.yml`) enforces a warning-count ratchet via `MAX_WARNINGS` — lower it when fixing findings; never raise it without justification.
 - **No-heap lint**: `scripts/check_no_heap.py --self-test` (also a ctest) rejects heap allocation in C++ sources under `src/`.

docs/cpp-core.md

@@ -1,8 +1,10 @@
 # C++20 core and allocation policy
 
-Status: phase 1 of an incremental port. The ownership and allocation
-primitives in `src/core/` exist and are tested; no program code uses them yet.
-All binaries are still built from the C sources.
+Status: phases 1-3 of an incremental port. Every module of `src/` is
+compiled as C++20 (the code is still the C design, converted mechanically).
+The ownership and allocation primitives in `src/core/` and the event layer
+in `src/event/` exist and are tested; the relay engines that will use them
+are the next phases.
 
 ## Goals
 
@@ -47,12 +49,28 @@ All binaries are still built from the C sources.
 | `intrusive_list.hpp` | `IntrusiveList`/`ListNode`; replaces `ss_list`. Nodes unlink themselves when destroyed. |
 | `fixed_heap.hpp` | Bounded size-class allocator to hand to libuv, c-ares and Mbed TLS through their allocator hooks. |
 
+## Event layer (`src/event/loop.hpp`)
+
+`Loop`, `Poll`, `Timer`, `Signal` and `Async` wrap libuv handles that are
+embedded by value in their owner and call a member function of the owner
+chosen at compile time (declare that member before the handle). A `Poll`
+owns its socket and carries both read and write interest on one libuv
+handle; a handle closed from another callback gets no further events.
+Handles close into a `CloseGroup`, which signals completion after the last
+asynchronous close, so pooled owners implement `AsyncClosable::begin_close`
+by closing their handles into one. `DeadlineList` drives the idle timeouts
+of many owners from one timer. The `ss_event` C API remains for the code
+that still uses it and is removed with its last caller.
+
 ## Phases
 
-1. Build and core (this change): C++ enabled in CMake, `src/core/`, unit tests
-   (`test_core_*`), the no-heap lint and the allocation audit.
-2. Mechanical C-to-C++ conversion, leaf modules first.
-3. Event loop wrapper over libuv.
+1. Build and core (done): C++ enabled in CMake, `src/core/`, unit tests
+   (`test_core_*`), the no-heap lint, the C++ runtime check and the
+   allocation audit.
+2. Mechanical C-to-C++ conversion, leaf modules first (done). Converted
+   modules still allocate like the C code; `scripts/no_heap_baseline.txt`
+   ratchets that down as they are rewritten.
+3. Event loop wrapper over libuv (done: `src/event/`, `test_event_cpp`).
 4. Crypto: cipher contexts held by value, dispatched with `std::variant`.
 5. Services: cache, lists, resolver, ACL and configuration on the new primitives.
 6. One TCP relay engine shared by ss-tunnel, ss-local, ss-redir and ss-server.

src/event/loop.hpp

@@ -0,0 +1,526 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+#ifndef SS_EVENT_LOOP_HPP
+#define SS_EVENT_LOOP_HPP
+
+// C++ event layer over libuv, replacing the ss_event C API module by module.
+//
+// Ownership rules
+//   * Every handle wrapper (Poll, Timer, Signal, Async) is embedded by value
+//     in the object that reacts to it, its Owner. Callbacks go straight to
+//     a member function of the Owner chosen at compile time; there is no
+//     container_of arithmetic and no per-handle heap state. The Owner must
+//     declare that member function before the handle member that names it.
+//   * A Poll owns its socket. The descriptor is closed in libuv's close
+//     callback, once libuv no longer polls it.
+//   * libuv closes handles asynchronously, so an Owner cannot be destroyed
+//     the moment it is done. It closes its handles into a CloseGroup, and
+//     the group reports completion once the last close callback has run.
+//     Owners kept in an ObjectPool model AsyncClosable by doing exactly
+//     that in begin_close(), so their slot is recycled only afterwards.
+//   * One poll handle per socket carries both read and write interest. A
+//     handle stopped or closed inside another callback gets no further
+//     events (libuv guarantees this), which is what the old ss_event
+//     generation snapshot emulated for shared per-fd watchers.
+
+#include <cstdint>
+#include <cstdlib>
+#include <optional>
+#include <type_traits>
+
+#include <uv.h>
+
+#include "core/intrusive_list.hpp"
+#include "core/pool.hpp"
+#include "core/result.hpp"
+#include "platform.h"
+
+namespace ss::event {
+
+// Readiness bits delivered to Poll owners.
+enum Events : int {
+    readable = 1,
+    writable = 2,
+    failed   = 4, // the socket reported an error; reading or writing will tell which
+};
+
+// What a callback did with its owner. Returning Flow::closed promises the
+// caller that the owner has started closing and must not be touched again.
+enum class [[nodiscard]] Flow : std::uint8_t { proceed, closed };
+
+class Loop {
+public:
+    Loop() noexcept = default;
+    Loop(const Loop &)            = delete;
+    Loop &operator=(const Loop &) = delete;
+
+    ~Loop()
+    {
+        if (initialised_ && !shutdown()) {
+            std::abort(); // a handle outlived the loop it belongs to
+        }
+    }
+
+    Status init() noexcept
+    {
+        if (initialised_) {
+            return fail(Errc::invalid);
+        }
+        int rc = uv_loop_init(&loop_);
+        if (rc != 0) {
+            return fail(Errc::io, -rc);
+        }
+        initialised_ = true;
+        return {};
+    }
+
+    // Run until stop() or until nothing is left to do.
+    void run() noexcept { uv_run(&loop_, UV_RUN_DEFAULT); }
+    // Poll once without blocking.
+    void run_nowait() noexcept { uv_run(&loop_, UV_RUN_NOWAIT); }
+    // Block for one round of events.
+    void run_once() noexcept { uv_run(&loop_, UV_RUN_ONCE); }
+    void stop() noexcept { uv_stop(&loop_); }
+
+    // Cached loop time in milliseconds.
+    std::uint64_t now() const noexcept { return uv_now(&loop_); }
+    void update_time() noexcept { uv_update_time(&loop_); }
+
+    uv_loop_t *native() noexcept { return &loop_; }
+
+    // Let pending close callbacks run and release the loop. Fails, leaving
+    // the loop usable, while handles are still open.
+    bool shutdown() noexcept
+    {
+        if (!initialised_) {
+            return true;
+        }
+        uv_run(&loop_, UV_RUN_NOWAIT);
+        if (uv_loop_close(&loop_) != 0) {
+            return false;
+        }
+        initialised_ = false;
+        return true;
+    }
+
+private:
+    uv_loop_t loop_{};
+    bool initialised_ = false;
+};
+
+// Collects the asynchronous closes of one owner's handles and invokes a
+// CloseCompletion after the last one. Usage inside begin_close():
+//
+//     closer_.arm(done);
+//     client_.close(closer_);
+//     timer_.close(closer_);
+//     closer_.seal();
+//
+// seal() completes immediately when nothing needed closing.
+class CloseGroup {
+public:
+    void arm(CloseCompletion done) noexcept
+    {
+        if (armed_) {
+            std::abort();
+        }
+        done_  = done;
+        armed_ = true;
+    }
+
+    void seal() noexcept
+    {
+        sealed_ = true;
+        maybe_finish();
+    }
+
+    // Called by handle wrappers.
+    void enter() noexcept { ++pending_; }
+    void leave() noexcept
+    {
+        if (pending_ == 0) {
+            std::abort();
+        }
+        --pending_;
+        maybe_finish();
+    }
+
+    bool closing() const noexcept { return armed_; }
+
+private:
+    void maybe_finish() noexcept
+    {
+        if (sealed_ && pending_ == 0 && armed_) {
+            armed_                = false;
+            CloseCompletion done = done_;
+            done(); // may destroy the owner, and with it this group
+        }
+    }
+
+    CloseCompletion done_;
+    std::uint32_t pending_ = 0;
+    bool armed_            = false;
+    bool sealed_           = false;
+};
+
+namespace detail {
+
+// State shared by the handle wrappers: the owner, the group to report a
+// finished close to, and whether the handle is live.
+template <class Handle>
+struct HandleBase {
+    HandleBase() noexcept = default;
+    HandleBase(const HandleBase &)            = delete;
+    HandleBase &operator=(const HandleBase &) = delete;
+
+    ~HandleBase()
+    {
+        if (live) {
+            std::abort(); // close() the handle and wait for it first
+        }
+    }
+
+    uv_handle_t *as_handle() noexcept { return reinterpret_cast<uv_handle_t *>(&handle); }
+
+    Handle handle{};
+    CloseGroup *group = nullptr;
+    bool live         = false;
+};
+
+} // namespace detail
+
+template <class Owner, Flow (Owner::*OnReady)(int events)>
+class Poll : private detail::HandleBase<uv_poll_t> {
+public:
+    // Start owning `fd`. Nothing is watched until want() asks for events.
+    Status open(Loop &loop, Owner &owner, ss_socket_t fd) noexcept
+    {
+        if (live) {
+            return fail(Errc::invalid);
+        }
+#ifdef _WIN32
+        int rc = uv_poll_init_socket(loop.native(), &handle, fd);
+#else
+        int rc = uv_poll_init(loop.native(), &handle, fd);
+#endif
+        if (rc != 0) {
+            return fail(Errc::io, -rc);
+        }
+        handle.data = this;
+        owner_      = &owner;
+        fd_         = fd;
+        interest_   = 0;
+        live        = true;
+        return {};
+    }
+
+    // Replace the set of events of interest (readable | writable); 0 stops
+    // polling without closing.
+    Status want(int events) noexcept
+    {
+        if (!live) {
+            return fail(Errc::closed);
+        }
+        events &= readable | writable;
+        if (events == interest_) {
+            return {};
+        }
+        int rc = 0;
+        if (events == 0) {
+            rc = uv_poll_stop(&handle);
+        } else {
+            int uv_events = ((events & readable) ? UV_READABLE : 0) | ((events & writable) ? UV_WRITABLE : 0);
+            rc = uv_poll_start(&handle, uv_events, &Poll::on_poll);
+        }
+        if (rc != 0) {
+            return fail(Errc::io, -rc);
+        }
+        interest_ = events;
+        return {};
+    }
+
+    int interest() const noexcept { return interest_; }
+    ss_socket_t fd() const noexcept { return fd_; }
+    bool is_open() const noexcept { return live; }
+
+    // Stop polling and close; the socket is closed once libuv lets go.
+    void close(CloseGroup &closer) noexcept
+    {
+        if (!live || group != nullptr) {
+            return;
+        }
+        group = &closer;
+        closer.enter();
+        uv_close(as_handle(), &Poll::on_closed);
+    }
+
+private:
+    static void on_poll(uv_poll_t *h, int status, int events) noexcept
+    {
+        auto *self = static_cast<Poll *>(h->data);
+        int ready  = ((events & UV_READABLE) ? readable : 0) | ((events & UV_WRITABLE) ? writable : 0);
+        if (status < 0) {
+            ready = readable | writable | failed;
+        }
+        (void)(self->owner_->*OnReady)(ready);
+    }
+
+    static void on_closed(uv_handle_t *h) noexcept
+    {
+        auto *self = static_cast<Poll *>(h->data);
+        ss_socket_close(self->fd_);
+        self->live           = false;
+        CloseGroup *closer   = self->group;
+        self->group          = nullptr;
+        closer->leave(); // may destroy self
+    }
+
+    Owner *owner_   = nullptr;
+    ss_socket_t fd_ = 0;
+    int interest_   = 0;
+};
+
+template <class Owner, void (Owner::*OnFire)()>
+class Timer : private detail::HandleBase<uv_timer_t> {
+public:
+    Status open(Loop &loop, Owner &owner) noexcept
+    {
+        if (live) {
+            return fail(Errc::invalid);
+        }
+        int rc = uv_timer_init(loop.native(), &handle);
+        if (rc != 0) {
+            return fail(Errc::io, -rc);
+        }
+        handle.data = this;
+        owner_      = &owner;
+        live        = true;
+        return {};
+    }
+
+    // Fire after `timeout_ms`, then every `repeat_ms` if non-zero.
+    void start(std::uint64_t timeout_ms, std::uint64_t repeat_ms = 0) noexcept
+    {
+        if (live) {
+            (void)uv_timer_start(&handle, &Timer::on_fire, timeout_ms, repeat_ms);
+        }
+    }
+    void stop() noexcept
+    {
+        if (live) {
+            (void)uv_timer_stop(&handle);
+        }
+    }
+    bool active() const noexcept { return live && uv_is_active(reinterpret_cast<const uv_handle_t *>(&handle)) != 0; }
+    bool is_open() const noexcept { return live; }
+
+    void close(CloseGroup &closer) noexcept
+    {
+        if (!live || group != nullptr) {
+            return;
+        }
+        group = &closer;
+        closer.enter();
+        uv_close(as_handle(), &Timer::on_closed);
+    }
+
+private:
+    static void on_fire(uv_timer_t *h) noexcept
+    {
+        auto *self = static_cast<Timer *>(h->data);
+        (self->owner_->*OnFire)();
+    }
+
+    static void on_closed(uv_handle_t *h) noexcept
+    {
+        auto *self         = static_cast<Timer *>(h->data);
+        self->live         = false;
+        CloseGroup *closer = self->group;
+        self->group        = nullptr;
+        closer->leave();
+    }
+
+    Owner *owner_ = nullptr;
+};
+
+template <class Owner, void (Owner::*OnSignal)(int signum)>
+class Signal : private detail::HandleBase<uv_signal_t> {
+public:
+    Status open(Loop &loop, Owner &owner, int signum) noexcept
+    {
+        if (live) {
+            return fail(Errc::invalid);
+        }
+        int rc = uv_signal_init(loop.native(), &handle);
+        if (rc != 0) {
+            return fail(Errc::io, -rc);
+        }
+        handle.data = this;
+        owner_      = &owner;
+        live        = true;
+        rc          = uv_signal_start(&handle, &Signal::on_signal, signum);
+        if (rc != 0) {
+            // The handle stays open (but idle) so the owner's normal close
+            // path releases it.
+            return fail(Errc::io, -rc);
+        }
+        return {};
+    }
+
+    bool is_open() const noexcept { return live; }
+
+    void close(CloseGroup &closer) noexcept
+    {
+        if (!live || group != nullptr) {
+            return;
+        }
+        group = &closer;
+        closer.enter();
+        uv_close(as_handle(), &Signal::on_closed);
+    }
+
+private:
+    static void on_signal(uv_signal_t *h, int signum) noexcept
+    {
+        auto *self = static_cast<Signal *>(h->data);
+        (self->owner_->*OnSignal)(signum);
+    }
+
+    static void on_closed(uv_handle_t *h) noexcept
+    {
+        auto *self         = static_cast<Signal *>(h->data);
+        self->live         = false;
+        CloseGroup *closer = self->group;
+        self->group        = nullptr;
+        closer->leave();
+    }
+
+    Owner *owner_ = nullptr;
+};
+
+// Wakes the loop from any thread; the callback runs on the loop thread.
+// Several send() calls before the loop wakes up coalesce into one callback.
+template <class Owner, void (Owner::*OnWake)()>
+class Async : private detail::HandleBase<uv_async_t> {
+public:
+    Status open(Loop &loop, Owner &owner) noexcept
+    {
+        if (live) {
+            return fail(Errc::invalid);
+        }
+        int rc = uv_async_init(loop.native(), &handle, &Async::on_wake);
+        if (rc != 0) {
+            return fail(Errc::io, -rc);
+        }
+        handle.data = this;
+        owner_      = &owner;
+        live        = true;
+        return {};
+    }
+
+    // Thread-safe and async-signal-safe.
+    void send() noexcept { (void)uv_async_send(&handle); }
+    bool is_open() const noexcept { return live; }
+
+    void close(CloseGroup &closer) noexcept
+    {
+        if (!live || group != nullptr) {
+            return;
+        }
+        group = &closer;
+        closer.enter();
+        uv_close(as_handle(), &Async::on_closed);
+    }
+
+private:
+    static void on_wake(uv_async_t *h) noexcept
+    {
+        auto *self = static_cast<Async *>(h->data);
+        (self->owner_->*OnWake)();
+    }
+
+    static void on_closed(uv_handle_t *h) noexcept
+    {
+        auto *self         = static_cast<Async *>(h->data);
+        self->live         = false;
+        CloseGroup *closer = self->group;
+        self->group        = nullptr;
+        closer->leave();
+    }
+
+    Owner *owner_ = nullptr;
+};
+
+// Idle timeouts for many objects that share one timeout value, driven by a
+// single timer. Because every entry gets the same timeout, refreshing an
+// entry is a move to the back of the list, and the front is always the next
+// one to expire.
+struct DeadlineTag;
+
+class Deadline : public ListNode<DeadlineTag> {
+public:
+    std::uint64_t deadline() const noexcept { return at_; }
+
+private:
+    template <class>
+    friend class DeadlineList;
+    std::uint64_t at_ = 0;
+};
+
+template <class T>
+class DeadlineList {
+    static_assert(std::is_base_of_v<Deadline, T>);
+
+public:
+    explicit DeadlineList(std::uint64_t timeout_ms = 0) noexcept : timeout_(timeout_ms) {}
+
+    void set_timeout(std::uint64_t timeout_ms) noexcept { timeout_ = timeout_ms; }
+    std::uint64_t timeout() const noexcept { return timeout_; }
+
+    // (Re)start the item's timeout from `now`.
+    void touch(T &item, std::uint64_t now) noexcept
+    {
+        static_cast<Deadline &>(item).at_ = now + timeout_;
+        if (static_cast<ListNode<DeadlineTag> &>(item).linked()) {
+            list_.move_to_back(item);
+        } else {
+            list_.push_back(item);
+        }
+    }
+
+    static void remove(T &item) noexcept { IntrusiveList<T, DeadlineTag>::remove(item); }
+
+    bool empty() const noexcept { return list_.empty(); }
+
+    // When the earliest entry expires, if any.
+    std::optional<std::uint64_t> next_due() noexcept
+    {
+        T *first = list_.front();
+        if (first == nullptr) {
+            return std::nullopt;
+        }
+        return static_cast<Deadline *>(first)->at_;
+    }
+
+    // Unlink every entry due at `now` and hand it to `on_expired`, which may
+    // destroy it.
+    template <class F>
+    std::size_t expire(std::uint64_t now, F &&on_expired)
+    {
+        std::size_t n = 0;
+        for (T *first = list_.front(); first != nullptr && static_cast<Deadline *>(first)->at_ <= now;
+             first    = list_.front()) {
+            remove(*first);
+            on_expired(*first);
+            ++n;
+        }
+        return n;
+    }
+
+private:
+    IntrusiveList<T, DeadlineTag> list_;
+    std::uint64_t timeout_;
+};
+
+} // namespace ss::event
+
+#endif

tests/CMakeLists.txt

@@ -63,6 +63,11 @@ foreach(core_test result pool buffer arena fixed_map ownership)
     ss_strict_warnings(test_core_${core_test})
 endforeach()
 
+# C++ event layer (src/event) over libuv.
+ss_add_test(test_event_cpp "test_event_cpp.cpp" "ss_uv")
+ss_strict_warnings(test_event_cpp)
+set_tests_properties(test_event_cpp PROPERTIES TIMEOUT 30)
+
 # C++ sources under src/ must not allocate on the heap (see the script).
 find_package(Python3 COMPONENTS Interpreter QUIET)
 if(Python3_Interpreter_FOUND)

tests/test_event_cpp.cpp

@@ -0,0 +1,307 @@
+/* SPDX-License-Identifier: GPL-3.0-or-later */
+// C++ event layer (src/event/loop.hpp): polling, close ordering, pooled
+// owners, timers, deadlines and cross-thread wakeups.
+#include <cstdint>
+#include <cstring>
+#include <optional>
+
+#include "event/loop.hpp"
+#include "test_helpers.h"
+#include "test_helpers.hpp"
+
+namespace ev = ss::event;
+
+namespace {
+
+ss_socket_t udp_socket(sockaddr_in &bound)
+{
+    ss_socket_t fd = socket(AF_INET, SOCK_DGRAM, 0);
+    SS_CHECK(fd != (ss_socket_t)-1);
+    sockaddr_in addr{};
+    addr.sin_family      = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    SS_CHECK(bind(fd, reinterpret_cast<sockaddr *>(&addr), sizeof(addr)) == 0);
+    socklen_t len = sizeof(bound);
+    SS_CHECK(getsockname(fd, reinterpret_cast<sockaddr *>(&bound), &len) == 0);
+    return fd;
+}
+
+void send_byte(const sockaddr_in &to)
+{
+    sockaddr_in unused{};
+    ss_socket_t s = udp_socket(unused);
+    char byte     = 'x';
+    SS_CHECK(sendto(s, &byte, 1, 0, reinterpret_cast<const sockaddr *>(&to), sizeof(to)) == 1);
+    ss_socket_close(s);
+}
+
+void drain(ss_socket_t fd)
+{
+    char buf[16];
+    (void)recv(fd, buf, sizeof(buf), 0);
+}
+
+// Closes its handles into a group and records when the group completes.
+struct Finished {
+    bool done = false;
+    static void complete(void *self, std::uint32_t) noexcept { static_cast<Finished *>(self)->done = true; }
+    ss::CloseCompletion completion() noexcept { return ss::CloseCompletion(&Finished::complete, this, 0); }
+};
+
+struct Reader {
+    ev::Flow on_ready(int ready);
+
+    ev::Loop *loop = nullptr;
+    ev::Poll<Reader, &Reader::on_ready> poll;
+    ev::CloseGroup closer;
+    Reader *victim = nullptr; // closed from inside our callback
+    Finished victim_closed;
+    int events = 0;
+};
+
+ev::Flow Reader::on_ready(int ready)
+{
+    ++events;
+    SS_CHECK(ready & ev::readable);
+    drain(poll.fd());
+    if (victim != nullptr) {
+        victim->closer.arm(victim_closed.completion());
+        victim->poll.close(victim->closer);
+        victim->closer.seal();
+        victim = nullptr;
+    }
+    loop->stop();
+    return ev::Flow::proceed;
+}
+
+void close_reader(ev::Loop &loop, Reader &r)
+{
+    Finished finished;
+    r.closer.arm(finished.completion());
+    r.poll.close(r.closer);
+    r.closer.seal();
+    while (!finished.done) {
+        loop.run_once();
+    }
+}
+
+void test_poll_readable()
+{
+    ev::Loop loop;
+    SS_CHECK(loop.init());
+    sockaddr_in addr{};
+    Reader r;
+    r.loop = &loop;
+    SS_CHECK(r.poll.open(loop, r, udp_socket(addr)));
+    SS_CHECK(r.poll.want(ev::readable));
+    send_byte(addr);
+    loop.run();
+    SS_CHECK(r.events == 1);
+    // Interest can be dropped without closing.
+    SS_CHECK(r.poll.want(0));
+    SS_CHECK(r.poll.interest() == 0);
+    // The loop refuses to shut down while a handle is still open.
+    SS_CHECK(!loop.shutdown());
+    close_reader(loop, r);
+    SS_CHECK(!r.poll.is_open());
+    SS_CHECK(loop.shutdown());
+}
+
+void test_close_peer_inside_callback()
+{
+    ev::Loop loop;
+    SS_CHECK(loop.init());
+    sockaddr_in a_addr{}, b_addr{};
+    Reader a, b;
+    a.loop = b.loop = &loop;
+    SS_CHECK(a.poll.open(loop, a, udp_socket(a_addr)));
+    SS_CHECK(b.poll.open(loop, b, udp_socket(b_addr)));
+    SS_CHECK(a.poll.want(ev::readable));
+    SS_CHECK(b.poll.want(ev::readable));
+    // Whichever fires first closes the other; the other must stay silent.
+    a.victim = &b;
+    b.victim = &a;
+    send_byte(a_addr);
+    send_byte(b_addr);
+    loop.run();
+    SS_CHECK(a.events + b.events == 1);
+    Reader &survivor = a.events == 1 ? a : b;
+    Reader &closed   = a.events == 1 ? b : a;
+    while (!survivor.victim_closed.done) {
+        loop.run_once();
+    }
+    SS_CHECK(!closed.poll.is_open());
+    loop.run_nowait();
+    SS_CHECK(closed.events == 0);
+    close_reader(loop, survivor);
+    SS_CHECK(loop.shutdown());
+}
+
+// A pooled connection-like owner: dropping its PoolPtr closes its handles,
+// and the slot only returns to the pool afterwards.
+struct Conn {
+    ev::Flow on_ready(int) { return ev::Flow::proceed; }
+    void on_timer() {}
+
+    ev::Poll<Conn, &Conn::on_ready> poll;
+    ev::Timer<Conn, &Conn::on_timer> timer;
+    ev::CloseGroup closer;
+
+    void begin_close(ss::CloseCompletion done) noexcept
+    {
+        closer.arm(done);
+        poll.close(closer);
+        timer.close(closer);
+        closer.seal();
+    }
+};
+
+void test_pooled_owner_close()
+{
+    static_assert(ss::AsyncClosable<Conn>);
+    ev::Loop loop;
+    SS_CHECK(loop.init());
+    ss::ObjectPool<Conn> pool;
+    SS_CHECK(pool.reserve(1));
+    sockaddr_in addr{};
+    ss_socket_t fd;
+    {
+        auto conn = pool.acquire();
+        SS_CHECK(conn);
+        fd = udp_socket(addr);
+        SS_CHECK((*conn)->poll.open(loop, **conn, fd));
+        SS_CHECK((*conn)->timer.open(loop, **conn));
+        (*conn)->timer.start(60000);
+        SS_CHECK((*conn)->poll.want(ev::readable));
+    }
+    // Ownership is gone; libuv still holds the handles.
+    SS_CHECK(pool.closing() == 1 && pool.available() == 0);
+    loop.run_nowait();
+    SS_CHECK(pool.closing() == 0 && pool.available() == 1);
+#ifndef _WIN32
+    // The socket was closed with the poll handle.
+    SS_CHECK(fcntl(fd, F_GETFD) == -1);
+#endif
+    (void)fd;
+    // An owner that never opened a handle completes synchronously.
+    {
+        auto idle = pool.acquire();
+        SS_CHECK(idle);
+    }
+    SS_CHECK(pool.available() == 1);
+    SS_CHECK(loop.shutdown());
+}
+
+struct Ticker {
+    void on_timer();
+
+    ev::Loop *loop = nullptr;
+    ev::Timer<Ticker, &Ticker::on_timer> timer;
+    ev::CloseGroup closer;
+    int fired = 0;
+};
+
+void Ticker::on_timer()
+{
+    if (++fired == 3) {
+        timer.stop();
+        loop->stop();
+    }
+}
+
+void test_timer_repeat()
+{
+    ev::Loop loop;
+    SS_CHECK(loop.init());
+    Ticker t;
+    t.loop = &loop;
+    SS_CHECK(t.timer.open(loop, t));
+    t.timer.start(1, 1);
+    SS_CHECK(t.timer.active());
+    loop.run();
+    SS_CHECK(t.fired == 3 && !t.timer.active());
+    Finished finished;
+    t.closer.arm(finished.completion());
+    t.timer.close(t.closer);
+    t.closer.seal();
+    loop.run();
+    SS_CHECK(finished.done && loop.shutdown());
+}
+
+struct Idle : ev::Deadline {
+    explicit Idle(int v) noexcept : id(v) {}
+    int id;
+};
+
+void test_deadlines()
+{
+    ev::DeadlineList<Idle> idle(100);
+    Idle a(1), b(2), c(3);
+    SS_CHECK(!idle.next_due());
+    idle.touch(a, 0);
+    idle.touch(b, 10);
+    idle.touch(c, 20);
+    SS_CHECK(idle.next_due() == std::optional<std::uint64_t>(100));
+    // Activity on `a` pushes it behind the others.
+    idle.touch(a, 50);
+    SS_CHECK(idle.next_due() == std::optional<std::uint64_t>(110) && a.deadline() == 150);
+    int order[3] = {};
+    int n        = 0;
+    SS_CHECK(idle.expire(120, [&](Idle &x) { order[n++] = x.id; }) == 2);
+    SS_CHECK(order[0] == 2 && order[1] == 3);
+    SS_CHECK(!b.linked() && a.linked());
+    ev::DeadlineList<Idle>::remove(a);
+    SS_CHECK(idle.empty());
+}
+
+struct Waker {
+    void on_wake()
+    {
+        ++woken;
+        loop->stop();
+    }
+
+    ev::Loop *loop = nullptr;
+    ev::Async<Waker, &Waker::on_wake> async;
+    ev::CloseGroup closer;
+    int woken = 0;
+};
+
+void wake_from_thread(void *arg)
+{
+    static_cast<Waker *>(arg)->async.send();
+}
+
+void test_async_cross_thread()
+{
+    ev::Loop loop;
+    SS_CHECK(loop.init());
+    Waker w;
+    w.loop = &loop;
+    SS_CHECK(w.async.open(loop, w));
+    uv_thread_t thread;
+    SS_CHECK(uv_thread_create(&thread, &wake_from_thread, &w) == 0);
+    loop.run();
+    SS_CHECK(uv_thread_join(&thread) == 0);
+    SS_CHECK(w.woken == 1);
+    Finished finished;
+    w.closer.arm(finished.completion());
+    w.async.close(w.closer);
+    w.closer.seal();
+    loop.run();
+    SS_CHECK(finished.done && loop.shutdown());
+}
+
+} // namespace
+
+int main()
+{
+    test_network_init();
+    SS_RUN(test_poll_readable);
+    SS_RUN(test_close_peer_inside_callback);
+    SS_RUN(test_pooled_owner_close);
+    SS_RUN(test_timer_repeat);
+    SS_RUN(test_deadlines);
+    SS_RUN(test_async_cross_thread);
+    return 0;
+}