From 0f052fefd1b323ffd40f8960c066e0bb5152dfdf Mon Sep 17 00:00:00 2001 From: Charlie <5764343+charlielye@users.noreply.github.com> Date: Wed, 12 Aug 2026 14:55:26 +0000 Subject: [PATCH] fix: IPC reactor delivered stale responses to recycled client slots and died of SIGPIPE --- ipc-runtime/cpp/ipc_runtime/ipc_server.hpp | 44 +- .../cpp/ipc_runtime/signal_handlers.cpp | 5 + .../cpp/ipc_runtime/signal_handlers.hpp | 2 + ipc-runtime/cpp/ipc_runtime/socket.test.cpp | 193 +++++++++ ipc-runtime/cpp/ipc_runtime/socket_server.cpp | 137 +++--- ipc-runtime/cpp/ipc_runtime/socket_server.hpp | 23 +- ipc-runtime/ts/src/uds_client.ts | 10 +- .../src/native/ipc_churn_correlation.test.ts | 402 ++++++++++++++++++ .../src/native/wsdb_sigpipe_death.test.ts | 86 ++++ 9 files changed, 817 insertions(+), 85 deletions(-) create mode 100644 yarn-project/world-state/src/native/ipc_churn_correlation.test.ts create mode 100644 yarn-project/world-state/src/native/wsdb_sigpipe_death.test.ts diff --git a/ipc-runtime/cpp/ipc_runtime/ipc_server.hpp b/ipc-runtime/cpp/ipc_runtime/ipc_server.hpp index e05863b17681..64efdb6283cd 100644 --- a/ipc-runtime/cpp/ipc_runtime/ipc_server.hpp +++ b/ipc-runtime/cpp/ipc_runtime/ipc_server.hpp @@ -102,6 +102,21 @@ class IpcServer { */ virtual bool has_pending_request() { return wait_for_data(0) >= 0; } + /** + * @brief Client ids whose connection ended since the last call. + * + * run_reactor() polls this each iteration and erases the per-connection + * reorder state for each returned id — garbage collection, plus late + * responses for an erased id are dropped instead of written to a dead fd. + * The socket transport never reuses client ids, so an id's state can never + * be inherited by a later connection. The default (transports that do not + * observe disconnects) returns nothing. NOTE: a transport that recycles ids + * (MPSC-SHM's are physical ring indices) must not adopt this hook as-is — + * erase-on-disconnect alone cannot stop a late response from landing in a + * recycled id's fresh state; it needs an occupancy guard on respond(). + */ + virtual std::vector drain_disconnected_clients() { return {}; } + /** * @brief Receive next message from a specific client * @@ -344,7 +359,20 @@ class IpcServer { } }; + // Reactor-only: garbage-collect the reorder state of connections that + // ended. Client ids are never reused, so this is purely reclamation — + // and once erased, a late respond() for the dead connection finds no + // entry and is dropped instead of being written to a dead fd. + auto drain_disconnects = [&]() { + for (int dead : drain_disconnected_clients()) { + std::lock_guard lock(mtx); + conns.erase(dead); + next_seq.erase(dead); + } + }; + while (!shutdown_requested_.load(std::memory_order_acquire)) { + drain_disconnects(); accept(); int client_id = wait_for_data_or_ready(100000000, have_ready); // 100ms shutdown backstop @@ -365,6 +393,13 @@ class IpcServer { release(client_id, request.size()); uint64_t seq = next_seq[client_id]++; + { + // Create the connection's reorder entry on the reactor thread; respond() + // only ever find()s, so a connection erased by drain_disconnects can + // never be resurrected by a late completion. + std::lock_guard lock(mtx); + conns.try_emplace(client_id); + } inflight.fetch_add(1, std::memory_order_relaxed); // respond(): invoked exactly once, possibly on another thread. Stash @@ -373,10 +408,15 @@ class IpcServer { // wake is never lost. Holds `buf` alive until invoked. Captures reactor // locals by reference, valid because run_reactor does not return until // inflight hits 0 (quiesce) and the final respond drives it there. + // A response for a connection that has since ended finds no entry + // (client ids are never reused) and is dropped. Respond respond = [this, client_id, seq, buf, &mtx, &conns, &inflight](std::vector response) { { std::lock_guard lock(mtx); - conns[client_id].stash.emplace(seq, std::move(response)); + auto it = conns.find(client_id); + if (it != conns.end()) { + it->second.stash.emplace(seq, std::move(response)); + } } inflight.fetch_sub(1, std::memory_order_release); notify(); @@ -390,9 +430,11 @@ class IpcServer { // (mtx, conns, inflight), so we must not unwind until every respond() has // fired. while (inflight.load(std::memory_order_acquire) > 0) { + drain_disconnects(); drain_and_send(); wait_for_data_or_ready(10000000, have_ready); } + drain_disconnects(); drain_and_send(); } diff --git a/ipc-runtime/cpp/ipc_runtime/signal_handlers.cpp b/ipc-runtime/cpp/ipc_runtime/signal_handlers.cpp index 52cf6aecd650..41ec1f9db100 100644 --- a/ipc-runtime/cpp/ipc_runtime/signal_handlers.cpp +++ b/ipc-runtime/cpp/ipc_runtime/signal_handlers.cpp @@ -137,6 +137,11 @@ void install_default_signal_handlers(IpcServer& server) (void)std::signal(SIGINT, graceful_shutdown_handler); (void)std::signal(SIGBUS, fatal_error_handler); (void)std::signal(SIGSEGV, fatal_error_handler); + // A client that disconnects with responses still in flight must produce + // EPIPE on the server's send(), never a process-killing SIGPIPE. send() + // already passes MSG_NOSIGNAL where available; this covers every other + // write to a peer-closed fd. + (void)std::signal(SIGPIPE, SIG_IGN); setup_parent_death_monitoring(); } diff --git a/ipc-runtime/cpp/ipc_runtime/signal_handlers.hpp b/ipc-runtime/cpp/ipc_runtime/signal_handlers.hpp index 83f0442d78a3..abd935e02c8c 100644 --- a/ipc-runtime/cpp/ipc_runtime/signal_handlers.hpp +++ b/ipc-runtime/cpp/ipc_runtime/signal_handlers.hpp @@ -8,6 +8,8 @@ * (graceful drain; the run() loop exits on its next poll iteration) * - SIGBUS / SIGSEGV → best-effort unlink of the server's socket/SHM * files (cached at install time) + _Exit(128 + sig) + * - SIGPIPE → SIG_IGN (a peer-closed fd yields EPIPE from write/send + * instead of killing the process) * - Parent-process death watch via prctl(PR_SET_PDEATHSIG) on Linux * and a kqueue NOTE_EXIT watcher on macOS — so spawn-and-forget * services die with their parent rather than turning into orphans. diff --git a/ipc-runtime/cpp/ipc_runtime/socket.test.cpp b/ipc-runtime/cpp/ipc_runtime/socket.test.cpp index c9e1cd48cc25..03fb9ceb0167 100644 --- a/ipc-runtime/cpp/ipc_runtime/socket.test.cpp +++ b/ipc-runtime/cpp/ipc_runtime/socket.test.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -268,4 +269,196 @@ TEST(SocketTest, ClientRejectsOversizedLengthPrefix) ::unlink(path.c_str()); } +// A connection that dies with a request still in flight leaves a late respond(). Client ids are +// never reused, so that response has nowhere valid to go — the reactor must drop it, and the next +// connection (a fresh id) must see only its own frames. Positional clients (the TS AsyncApi) +// depend on this: a single leaked frame shifts every subsequent response onto the wrong caller. +// +// Protocol: request = [tag]; response = [tag, client_id]. Tag 'A' defers its response behind a +// test-controlled gate (scripted completion order, no sleeps-as-sync); any other tag responds +// inline. The client_id echo pins the never-reused-id invariant directly. +TEST(SocketTest, ReactorDropsStaleResponsesAndNeverReusesIds) +{ + std::string path = test_socket_path("staleresp"); + auto server = IpcServer::create_socket(path, 4); + ASSERT_TRUE(server->listen()); + + std::mutex gate_m; + std::condition_variable gate_cv; + bool gate_open = false; + auto release_gate = [&] { + { + std::lock_guard lock(gate_m); + gate_open = true; + } + gate_cv.notify_all(); + }; + + std::promise a_request_seen; // fulfilled with the client_id that sent 'A' + TestPool pool(2); + + std::thread server_thread([&] { + server->run_reactor([&](int client_id, std::span req, IpcServer::Respond respond) { + uint8_t tag = req[0]; + if (tag == 'A') { + a_request_seen.set_value(client_id); + pool.enqueue([&gate_m, &gate_cv, &gate_open, client_id, respond = std::move(respond)]() mutable { + std::unique_lock lock(gate_m); + gate_cv.wait(lock, [&] { return gate_open; }); + respond({ 'A', static_cast(client_id) }); + }); + } else { + respond({ tag, static_cast(client_id) }); + } + }); + }); + + // Connection A: send 'A' (its response is now in flight behind the gate), then vanish. + auto client_a = IpcClient::create_socket(path); + ASSERT_TRUE(client_a->connect()); + uint8_t tag_a = 'A'; + ASSERT_TRUE(client_a->send(&tag_a, 1, 1'000'000'000ULL)); + auto a_seen = a_request_seen.get_future(); + ASSERT_EQ(a_seen.wait_for(std::chrono::seconds(2)), std::future_status::ready) << "server never saw A's request"; + int a_id = a_seen.get(); + client_a->close(); + + // Let the reactor observe A's EOF before B connects — the window in which A's id would be + // freed for reuse if ids were recycled. + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + // Connection B: sends 'B' (inline response), THEN A's zombie completes. + auto client_b = IpcClient::create_socket(path); + ASSERT_TRUE(client_b->connect()); + uint8_t tag_b = 'B'; + ASSERT_TRUE(client_b->send(&tag_b, 1, 1'000'000'000ULL)); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); // let B's response reach the stash + release_gate(); + + auto first = client_b->receive(2'000'000'000ULL); + ASSERT_EQ(first.size(), 2U) << "no response frame reached connection B"; + uint8_t first_tag = first[0]; + uint8_t first_id = first[1]; + client_b->release(first.size()); + + EXPECT_EQ(first_tag, 'B') << "connection B's first response frame carries the dead connection A's payload " + "(tag '" + << static_cast(first_tag) << "', id " << int(first_id) + << ") — a stale response was delivered across connections"; + EXPECT_NE(int(first_id), a_id) << "client id was reused across connections"; + + // And there must be exactly one frame: a leaked zombie shifts B's real response into a + // second frame (which a positional client would hand to the NEXT caller). + auto extra = client_b->receive(300'000'000ULL); + EXPECT_TRUE(extra.empty()) << "extra frame leaked to connection B (tag '" + << static_cast(extra.empty() ? '?' : extra[0]) << "')"; + if (!extra.empty()) { + client_b->release(extra.size()); + } + + client_b->close(); + server->request_shutdown(); + server_thread.join(); + server->close(); +} + +// Control for the scenario above: when the first connection's response completes and is read +// BEFORE it disconnects, the second connection (fresh id) sees exactly its own response. Pins the +// invariant and validates the harness. +TEST(SocketTest, ReactorSequentialConnectionsAreIndependent) +{ + std::string path = test_socket_path("seq_conns"); + auto server = IpcServer::create_socket(path, 4); + ASSERT_TRUE(server->listen()); + + std::thread server_thread([&] { + server->run_reactor([&](int client_id, std::span req, IpcServer::Respond respond) { + respond({ req[0], static_cast(client_id) }); + }); + }); + + auto client_a = IpcClient::create_socket(path); + ASSERT_TRUE(client_a->connect()); + uint8_t tag_a = 'A'; + ASSERT_TRUE(client_a->send(&tag_a, 1, 1'000'000'000ULL)); + auto resp_a = client_a->receive(2'000'000'000ULL); + ASSERT_EQ(resp_a.size(), 2U); + EXPECT_EQ(resp_a[0], 'A'); + uint8_t a_id = resp_a[1]; + client_a->release(resp_a.size()); + client_a->close(); + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + auto client_b = IpcClient::create_socket(path); + ASSERT_TRUE(client_b->connect()); + uint8_t tag_b = 'B'; + ASSERT_TRUE(client_b->send(&tag_b, 1, 1'000'000'000ULL)); + auto resp_b = client_b->receive(2'000'000'000ULL); + ASSERT_EQ(resp_b.size(), 2U); + EXPECT_EQ(resp_b[0], 'B'); + EXPECT_NE(resp_b[1], a_id) << "client id was reused across connections"; + client_b->release(resp_b.size()); + auto extra = client_b->receive(300'000'000ULL); + EXPECT_TRUE(extra.empty()); + + client_b->close(); + server->request_shutdown(); + server_thread.join(); + server->close(); +} + +// The reactor must survive a client that disconnects with responses still in flight: the +// reactor drops the late responses (their connection's state is gone) instead of writing them +// to the dead fd, and any +// write that does hit a peer-closed fd yields EPIPE (MSG_NOSIGNAL / SO_NOSIGPIPE), never a +// process-killing SIGPIPE. NOTE: an in-process peer-closed write can be absorbed by kernel +// buffering, so this test alone cannot prove SIGPIPE immunity — the cross-process guard is +// yarn-project/world-state's wsdb churn test, where the server lives in its own process. +TEST(SocketTest, ReactorSurvivesResponseToDeadClient) +{ + std::string path = test_socket_path("sigpipe"); + auto server = IpcServer::create_socket(path, 4); + ASSERT_TRUE(server->listen()); + + std::mutex gate_m; + std::condition_variable gate_cv; + bool gate_open = false; + TestPool pool(2); + + std::thread server_thread([&] { + server->run_reactor([&](int, std::span req, IpcServer::Respond respond) { + std::vector big(64 * 1024, req[0]); // big frames: force multiple send() calls + pool.enqueue([&gate_m, &gate_cv, &gate_open, big = std::move(big), respond = std::move(respond)]() mutable { + std::unique_lock lock(gate_m); + gate_cv.wait(lock, [&] { return gate_open; }); + respond(std::move(big)); + }); + }); + }); + + // Pipeline several requests, then vanish without reading anything. + auto client = IpcClient::create_socket(path); + ASSERT_TRUE(client->connect()); + for (uint8_t t = 0; t < 4; t++) { + ASSERT_TRUE(client->send(&t, 1, 1'000'000'000ULL)); + } + std::this_thread::sleep_for(std::chrono::milliseconds(100)); // let the reactor ingest all four + client->close(); + + // Release all four responses; the reactor must drop or fail them without dying. + { + std::lock_guard lock(gate_m); + gate_open = true; + } + gate_cv.notify_all(); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + + // If we are still alive, the server survived its client's mid-flight death. + server->request_shutdown(); + server_thread.join(); + server->close(); + SUCCEED(); +} + } // namespace diff --git a/ipc-runtime/cpp/ipc_runtime/socket_server.cpp b/ipc-runtime/cpp/ipc_runtime/socket_server.cpp index 2d5995fa606d..a39148d1695d 100644 --- a/ipc-runtime/cpp/ipc_runtime/socket_server.cpp +++ b/ipc-runtime/cpp/ipc_runtime/socket_server.cpp @@ -48,13 +48,14 @@ void SocketServer::close() void SocketServer::close_internal() { // Close all client connections - for (int fd : client_fds_) { + for (const auto& [client_id, fd] : client_fds_) { if (fd >= 0) { ::close(fd); } } client_fds_.clear(); fd_to_client_id_.clear(); + recv_buffers_.clear(); num_clients_ = 0; if (wake_read_fd_ >= 0) { @@ -80,19 +81,6 @@ void SocketServer::close_internal() ::unlink(socket_path_.c_str()); } -int SocketServer::find_free_slot() -{ - // Look for existing free slot - for (size_t i = 0; i < client_fds_.size(); i++) { - if (client_fds_[i] == -1) { - return static_cast(i); - } - } - - // No free slot found, allocate new one at end - return static_cast(client_fds_.size()); -} - bool SocketServer::setup_wake_pipe() { int fds[2]; @@ -153,8 +141,8 @@ void SocketServer::notify() bool SocketServer::send(int client_id, const void* data, size_t len) { - if (client_id < 0 || static_cast(client_id) >= client_fds_.size() || - client_fds_[static_cast(client_id)] < 0) { + auto fd_it = client_fds_.find(client_id); + if (fd_it == client_fds_.end()) { errno = EINVAL; return false; } @@ -164,18 +152,27 @@ bool SocketServer::send(int client_id, const void* data, size_t len) return false; } - int fd = client_fds_[static_cast(client_id)]; + int fd = fd_it->second; // Send length prefix (4 bytes) then message data, looping on partial // writes — a short write after the prefix would permanently desync the // stream for this connection. + // + // MSG_NOSIGNAL: a peer that closed with responses still in flight must + // yield EPIPE here, not a process-killing SIGPIPE. macOS has no + // MSG_NOSIGNAL; accept() sets SO_NOSIGPIPE on the fd instead. +#ifdef MSG_NOSIGNAL + constexpr int send_flags = MSG_NOSIGNAL; +#else + constexpr int send_flags = 0; +#endif auto msg_len = static_cast(len); const uint8_t* parts[2] = { reinterpret_cast(&msg_len), static_cast(data) }; size_t part_lens[2] = { sizeof(msg_len), len }; for (int part = 0; part < 2; part++) { size_t total_sent = 0; while (total_sent < part_lens[part]) { - ssize_t n = ::send(fd, parts[part] + total_sent, part_lens[part] - total_sent, 0); + ssize_t n = ::send(fd, parts[part] + total_sent, part_lens[part] - total_sent, send_flags); if (n < 0) { if (errno == EINTR) { continue; // Interrupted, retry @@ -201,18 +198,16 @@ void SocketServer::release(int client_id, size_t message_size) std::span SocketServer::receive(int client_id) { - if (client_id < 0 || static_cast(client_id) >= client_fds_.size() || - client_fds_[static_cast(client_id)] < 0) { + auto fd_it = client_fds_.find(client_id); + if (fd_it == client_fds_.end()) { return {}; } - int fd = client_fds_[static_cast(client_id)]; - const auto client_idx = static_cast(client_id); - - // Ensure buffers are sized for this client - if (client_idx >= recv_buffers_.size()) { - recv_buffers_.resize(client_idx + 1); - } + int fd = fd_it->second; + // Default-constructs this client's buffer on first use. disconnect_client() erases the + // map entry and invalidates this reference — every disconnect path below returns + // immediately without touching the buffer again. + std::vector& recv_buffer = recv_buffers_[client_id]; // Read length prefix (4 bytes) - must loop until all bytes received (MSG_WAITALL unreliable on macOS) uint32_t msg_len = 0; @@ -241,18 +236,17 @@ std::span SocketServer::receive(int client_id) // Resize buffer if needed to fit length prefix + message size_t total_size = sizeof(uint32_t) + msg_len; - if (recv_buffers_[client_idx].size() < total_size) { - recv_buffers_[client_idx].resize(total_size); + if (recv_buffer.size() < total_size) { + recv_buffer.resize(total_size); } // Store length prefix in buffer - std::memcpy(recv_buffers_[client_idx].data(), &msg_len, sizeof(uint32_t)); + std::memcpy(recv_buffer.data(), &msg_len, sizeof(uint32_t)); // Read message data - must loop until all bytes received (MSG_WAITALL unreliable on macOS) total_read = 0; while (total_read < msg_len) { - ssize_t n = - ::recv(fd, recv_buffers_[client_idx].data() + sizeof(uint32_t) + total_read, msg_len - total_read, 0); + ssize_t n = ::recv(fd, recv_buffer.data() + sizeof(uint32_t) + total_read, msg_len - total_read, 0); if (n < 0) { if (errno == EINTR) { continue; // Interrupted, retry @@ -268,7 +262,7 @@ std::span SocketServer::receive(int client_id) total_read += static_cast(n); } - return std::span(recv_buffers_[client_idx].data() + sizeof(uint32_t), msg_len); + return std::span(recv_buffer.data() + sizeof(uint32_t), msg_len); } #ifdef __APPLE__ @@ -388,15 +382,16 @@ int SocketServer::accept() fcntl(client_fd, F_SETFL, flags & ~O_NONBLOCK); } - // Find free slot (or allocate new one) - int client_id = find_free_slot(); +#ifdef SO_NOSIGPIPE + // No MSG_NOSIGNAL on macOS: suppress SIGPIPE per-fd so a send() to a + // disconnected peer yields EPIPE instead of killing the process. + int nosigpipe = 1; + setsockopt(client_fd, SOL_SOCKET, SO_NOSIGPIPE, &nosigpipe, sizeof(nosigpipe)); +#endif - // Store client fd - const auto client_id_unsigned = static_cast(client_id); - if (client_id_unsigned >= client_fds_.size()) { - client_fds_.resize(client_id_unsigned + 1, -1); - } - client_fds_[static_cast(client_id)] = client_fd; + // Fresh, never-reused connection id. + int client_id = next_client_id_++; + client_fds_[client_id] = client_fd; fd_to_client_id_[client_fd] = client_id; num_clients_++; @@ -468,23 +463,24 @@ int SocketServer::wait_for_data(uint64_t timeout_ns) void SocketServer::disconnect_client(int client_id) { - if (client_id < 0 || static_cast(client_id) >= client_fds_.size()) { + auto fd_it = client_fds_.find(client_id); + if (fd_it == client_fds_.end()) { return; } - int fd = client_fds_[static_cast(client_id)]; - if (fd >= 0) { - // For kqueue, we don't need explicit deletion - closing the fd removes it automatically - // But we can explicitly remove it for clarity - struct kevent ev; - EV_SET(&ev, fd, EVFILT_READ, EV_DELETE, 0, 0, nullptr); - kevent(fd_, &ev, 1, nullptr, 0, nullptr); - - ::close(fd); - fd_to_client_id_.erase(fd); - client_fds_[static_cast(client_id)] = -1; - num_clients_--; - } + int fd = fd_it->second; + // For kqueue, we don't need explicit deletion - closing the fd removes it automatically + // But we can explicitly remove it for clarity + struct kevent ev; + EV_SET(&ev, fd, EVFILT_READ, EV_DELETE, 0, 0, nullptr); + kevent(fd_, &ev, 1, nullptr, 0, nullptr); + + ::close(fd); + fd_to_client_id_.erase(fd); + client_fds_.erase(fd_it); + recv_buffers_.erase(client_id); + disconnected_clients_.push_back(client_id); + num_clients_--; } #else @@ -606,15 +602,9 @@ int SocketServer::accept() fcntl(client_fd, F_SETFL, flags & ~O_NONBLOCK); } - // Find free slot (or allocate new one) - int client_id = find_free_slot(); - - // Store client fd - const auto client_id_unsigned = static_cast(client_id); - if (client_id_unsigned >= client_fds_.size()) { - client_fds_.resize(client_id_unsigned + 1, -1); - } - client_fds_[static_cast(client_id)] = client_fd; + // Fresh, never-reused connection id. + int client_id = next_client_id_++; + client_fds_[client_id] = client_fd; fd_to_client_id_[client_fd] = client_id; num_clients_++; @@ -680,18 +670,19 @@ int SocketServer::wait_for_data(uint64_t timeout_ns) void SocketServer::disconnect_client(int client_id) { - if (client_id < 0 || static_cast(client_id) >= client_fds_.size()) { + auto fd_it = client_fds_.find(client_id); + if (fd_it == client_fds_.end()) { return; } - int fd = client_fds_[static_cast(client_id)]; - if (fd >= 0) { - epoll_ctl(fd_, EPOLL_CTL_DEL, fd, nullptr); - ::close(fd); - fd_to_client_id_.erase(fd); - client_fds_[static_cast(client_id)] = -1; - num_clients_--; - } + int fd = fd_it->second; + epoll_ctl(fd_, EPOLL_CTL_DEL, fd, nullptr); + ::close(fd); + fd_to_client_id_.erase(fd); + client_fds_.erase(fd_it); + recv_buffers_.erase(client_id); + disconnected_clients_.push_back(client_id); + num_clients_--; } #endif diff --git a/ipc-runtime/cpp/ipc_runtime/socket_server.hpp b/ipc-runtime/cpp/ipc_runtime/socket_server.hpp index b1e1bb1effbe..67eecf2c75ba 100644 --- a/ipc-runtime/cpp/ipc_runtime/socket_server.hpp +++ b/ipc-runtime/cpp/ipc_runtime/socket_server.hpp @@ -6,6 +6,7 @@ #include #include #include +#include #include namespace ipc { @@ -47,10 +48,12 @@ class SocketServer : public IpcServer { return CleanupPaths{ .unlink_paths = { socket_path_ }, .shm_unlink_names = {} }; } + // Reactor-thread only, like disconnect_client() (which records the ids). + std::vector drain_disconnected_clients() override { return std::exchange(disconnected_clients_, {}); } + private: void close_internal(); void disconnect_client(int client_id); - int find_free_slot(); // Create the self-pipe and register its read end with the epoll/kqueue // instance. Returns false on failure. Called from listen(). bool setup_wake_pipe(); @@ -60,12 +63,18 @@ class SocketServer : public IpcServer { std::string socket_path_; int initial_max_clients_; int listen_fd_ = -1; - int fd_ = -1; // kqueue or epoll fd - int wake_read_fd_ = -1; // self-pipe read end (in the event set) - int wake_write_fd_ = -1; // self-pipe write end (poked by notify()) - std::vector client_fds_; // client_id -> fd - std::unordered_map fd_to_client_id_; // fd -> client_id (for fast lookup) - std::vector> recv_buffers_; // client_id -> recv buffer + int fd_ = -1; // kqueue or epoll fd + int wake_read_fd_ = -1; // self-pipe read end (in the event set) + int wake_write_fd_ = -1; // self-pipe write end (poked by notify()) + // Client ids are monotonic and never reused: a connection's identity must not be + // inheritable by a later connection, or state addressed to a dead client (a late + // handler response, reorder bookkeeping) could reach its successor. Contrast with + // the SHM transport, whose ids are physical ring indices and must recycle. + int next_client_id_ = 0; + std::unordered_map client_fds_; // client_id -> fd + std::unordered_map fd_to_client_id_; // fd -> client_id (for fast lookup) + std::unordered_map> recv_buffers_; // client_id -> recv buffer + std::vector disconnected_clients_; // ids closed since the last drain (reactor thread only) int num_clients_ = 0; }; diff --git a/ipc-runtime/ts/src/uds_client.ts b/ipc-runtime/ts/src/uds_client.ts index 9127704f5d0a..004e968166d2 100644 --- a/ipc-runtime/ts/src/uds_client.ts +++ b/ipc-runtime/ts/src/uds_client.ts @@ -144,11 +144,12 @@ export class UdsIpcClient implements IpcClientAsync { /** * Connect to `socketPath`, retrying "server not ready" errors until * `timeoutMs` elapses: ENOENT (socket file not created yet), ECONNREFUSED - * (the window between the server's bind() and listen()), and EAGAIN (Linux + * (the window between the server's bind() and listen()), EAGAIN (Linux * reports this for a UDS connect when the accept backlog is momentarily - * full). Other errors fail immediately. Each attempt is also capped at the - * remaining budget, so a bound-but-never-accepting server cannot hang the - * connect past the deadline. + * full), and ECONNRESET (a connect racing the server's accept loop under + * connection churn). Other errors fail immediately. Each attempt is also + * capped at the remaining budget, so a bound-but-never-accepting server + * cannot hang the connect past the deadline. */ async function connectWithRetry( socketPath: string, @@ -166,6 +167,7 @@ async function connectWithRetry( const code = (err as NodeJS.ErrnoException).code; if ( code !== "ECONNREFUSED" && + code !== "ECONNRESET" && code !== "ENOENT" && code !== "ETIMEDOUT" && code !== "EAGAIN" diff --git a/yarn-project/world-state/src/native/ipc_churn_correlation.test.ts b/yarn-project/world-state/src/native/ipc_churn_correlation.test.ts new file mode 100644 index 000000000000..c3ef34b8aa93 --- /dev/null +++ b/yarn-project/world-state/src/native/ipc_churn_correlation.test.ts @@ -0,0 +1,402 @@ +import { Fr } from '@aztec/foundation/curves/bn254'; +import { type IpcClientAsync, UdsIpcClient, createNapiShmAsyncClient } from '@aztec/ipc-runtime'; +import { MerkleTreeId } from '@aztec/stdlib/trees'; +import { AsyncApi } from '@aztec/wsdb'; + +import { jest } from '@jest/globals'; + +import { NativeWorldStateService } from './native_world_state.js'; + +// Load tests for the wsdb IPC response-correlation invariant: every response a client resolves must +// be the response to ITS OWN request. The TS client correlates positionally (no request IDs), so any +// server-side misordering — reactor reorder stash, per-fork scheduler, or slot reuse across +// disconnects — surfaces here as a caller receiving another caller's payload. +// +// Detection design: connections must never share observable state, or a cross-connection swap of +// same-type responses is invisible (identical requests have identical answers). Each connection +// plants a private fork with leaves derived from its own seed, so ANY frame delivered to the wrong +// connection produces a visibly wrong value — wrong index, wrong root, wrong size — not just a +// wrong-type decode. +// +// Legs: +// C: single connection, sequential (sanity — must always pass). +// A: single connection, heavy pipelined contention (per-fork write serialisation → out-of-order +// completions feed the reorder stash) — models the long-lived world-state client. +// D: multi-connection read/write soak, NO connection churn. Readers verify recorded roots and +// exact leaf indices on their private forks; writers pipeline append→read-after-write with +// exact-index asserts. Independent of the disconnect-cleanup fixes — any failure here is a +// distinct IPC/reorder/scheduler bug. +// B: leg D's workload plus mid-flight churn — clients destroyed with calls in flight and +// replaced, recycling server slots. Guards the reactor's disconnect cleanup and the server's +// survival of mid-flight death. +// +// WSDB_SOAK_MS extends the D/B legs for grinding sessions (default a few seconds for CI). + +const SOAK_MS = Number(process.env.WSDB_SOAK_MS ?? 4_000); +jest.setTimeout(Math.max(300_000, SOAK_MS * 4)); + +const TRANSPORT = (process.env.WSDB_TRANSPORT === 'shm' ? 'shm' : 'uds') as 'shm' | 'uds'; +// Typed as plain number: the generated AsyncApi carries tree ids as numbers, so enum-typed +// comparisons against response fields trip no-unsafe-enum-comparison. +const TREE: number = MerkleTreeId.NOTE_HASH_TREE; +const N = 64; +/** Private leaves planted per connection identity. */ +const M = 32; + +/** Per-connection private state: a fork only this logical client touches. */ +interface ConnIdentity { + forkId: number; + rev: { forkId: number; blockNumber: number; includeUncommitted: boolean }; + /** Leaves unique to this identity (and disjoint from every other identity and the seed fork). */ + leaves: Uint8Array[]; + /** Tree size before this identity's leaves were appended — leaf i lives at baseSize + i. */ + baseSize: number; + /** Tree size including all writes so far (advanced by writer rounds). */ + size: number; + /** Root recorded after planting; stable until the identity writes again. */ + root: Uint8Array; +} + +/** Unique leaf value: disjoint across identities, write rounds, and the shared seed fork. */ +const identityLeaf = (seed: number, i: number): Uint8Array => + new Fr((BigInt(0xc0ffee) << 64n) + (BigInt(seed) << 32n) + BigInt(i) + 1n).toBuffer(); + +/** Progress line every ~15s so long soaks show liveness and rate instead of going dark. */ +const makeHeartbeat = (leg: string) => { + const start = Date.now(); + let lastLog = start; + return (rounds: number) => { + const now = Date.now(); + if (now - lastLog >= 15_000) { + lastLog = now; + const elapsedS = (now - start) / 1000; + // eslint-disable-next-line no-console + console.log( + `[soak ${leg}] ${new Date(now).toISOString()} rounds=${rounds} ` + + `elapsed=${elapsedS.toFixed(0)}s rate=${(rounds / elapsedS).toFixed(1)}/s`, + ); + } + }; +}; + +describe(`wsdb IPC churn/contention correlation (transport=${TRANSPORT})`, () => { + let ws: NativeWorldStateService; + let ipcPath: string | undefined; + let leaves: Fr[]; + // The generated AsyncApi is bytes-in/bytes-out (wire `Fr = Uint8Array`); the facade normally + // converts, but these tests drive AsyncApi directly, so pass raw 32-byte buffers. + let leafBytes: Uint8Array[]; + /** A shared fork seeded with `leaves`; read with includeUncommitted. */ + let seedForkId: number; + let seedRevision: { forkId: number; blockNumber: number; includeUncommitted: boolean }; + + const makeClient = async (clientId: number): Promise<{ api: AsyncApi; backend: IpcClientAsync }> => { + const backend: IpcClientAsync = + TRANSPORT === 'shm' + ? createNapiShmAsyncClient(ipcPath!.replace(/\.shm$/, ''), { clientId }) + : await UdsIpcClient.connect(ipcPath!, { connectTimeoutMs: 5_000 }); + return { api: new AsyncApi(backend), backend }; + }; + + /** Create a private fork and plant this identity's unique leaves, recording size and root. */ + const plantIdentity = async (api: AsyncApi, seed: number): Promise => { + const forkId = (await api.createFork({ latest: true, blockNumber: 0 })).forkId; + const rev = { forkId, blockNumber: 0xffffffff, includeUncommitted: true }; + const before = await api.getTreeInfo({ treeId: TREE, revision: rev }); + const baseSize = Number(before.size); + const idLeaves = Array.from({ length: M }, (_, i) => identityLeaf(seed, i)); + await api.appendLeaves({ treeId: TREE, leaves: idLeaves, forkId }); + const after = await api.getTreeInfo({ treeId: TREE, revision: rev }); + if (Number(after.size) !== baseSize + M) { + throw new Error(`plant(seed=${seed}): size ${after.size} != ${baseSize + M}`); + } + return { forkId, rev, leaves: idLeaves, baseSize, size: baseSize + M, root: after.root }; + }; + + /** + * Pipelined reads whose responses each prove they belong to this identity's own requests: exact + * leaf indices on the private fork, and (when the identity has not written since planting) the + * recorded root. A frame swapped from any other connection fails these by value. + */ + const identityChecks = (api: AsyncApi, id: ConnIdentity, tag: string, opts?: { checkRoot?: boolean }) => { + const checks: Promise[] = []; + for (let i = 0; i < M; i++) { + const expectIdx = id.baseSize + i; + checks.push( + api.findLeafIndices({ treeId: TREE, revision: id.rev, leaves: [id.leaves[i]], startIndex: 0 }).then(r => { + const got = r.indices?.[0]; + if (got === null || got === undefined || Number(got) !== expectIdx) { + throw new Error(`${tag}: findLeafIndices(own leaf ${i}) returned ${String(got)}, expected ${expectIdx}`); + } + }), + ); + checks.push( + api.getSiblingPath({ treeId: TREE, revision: id.rev, leafIndex: expectIdx }).then(r => { + if (!Array.isArray(r.path) || r.path.length === 0) { + throw new Error(`${tag}: getSiblingPath(${expectIdx}) returned empty/invalid path`); + } + }), + ); + checks.push( + api.getTreeInfo({ treeId: TREE, revision: id.rev }).then(r => { + if (r.treeId !== TREE || Number(r.size) < id.baseSize + M) { + throw new Error(`${tag}: getTreeInfo returned treeId=${r.treeId} size=${r.size} (min ${id.baseSize + M})`); + } + if (opts?.checkRoot && Buffer.compare(r.root, id.root) !== 0) { + throw new Error(`${tag}: getTreeInfo root mismatch — response belongs to another fork/connection`); + } + }), + ); + checks.push( + api.getStateReference({ revision: id.rev }).then(r => { + if (!Array.isArray(r.state) || r.state.length === 0 || r.state.some(t => t.size === undefined)) { + throw new Error(`${tag}: getStateReference malformed: ${JSON.stringify(r).slice(0, 160)}`); + } + }), + ); + } + return checks; + }; + + /** + * One writer round: pipeline an append of a fresh unique leaf and, behind it on the same + * connection, a read of that leaf — the per-fork scheduler must order the read after the write, + * and the leaf must land at exactly the pre-append size. Also refreshes the identity's root. + */ + const writerRound = async (api: AsyncApi, id: ConnIdentity, seed: number, round: number, tag: string) => { + const leaf = identityLeaf(seed, M + round); + const expectIdx = id.size; + await Promise.all([ + api.appendLeaves({ treeId: TREE, leaves: [leaf], forkId: id.forkId }), + api.findLeafIndices({ treeId: TREE, revision: id.rev, leaves: [leaf], startIndex: 0 }).then(r => { + const got = r.indices?.[0]; + if (got === null || got === undefined || Number(got) !== expectIdx) { + throw new Error( + `${tag}: read-after-write of round-${round} leaf returned ${String(got)}, expected ${expectIdx}`, + ); + } + }), + ...identityChecks(api, id, tag), + ]); + id.size++; + id.root = (await api.getTreeInfo({ treeId: TREE, revision: id.rev })).root; + }; + + beforeAll(async () => { + ws = await NativeWorldStateService.tmp(); + try { + ipcPath = ws.getIpcPath(); + } catch { + ipcPath = undefined; // in-process build — no IPC to exercise + return; + } + leaves = Array.from({ length: N }, (_, i) => new Fr(BigInt(i) * 0x1_0000_0001n + 1n)); + leafBytes = leaves.map(l => l.toBuffer()); + + // Seed a shared fork (forks are shared world-state objects, not per-connection). Legs A/C use + // it; the multi-connection legs deliberately do NOT — see the detection-design note above. + const seeder = await makeClient(1); + try { + seedForkId = (await seeder.api.createFork({ latest: true, blockNumber: 0 })).forkId; + seedRevision = { forkId: seedForkId, blockNumber: 0xffffffff, includeUncommitted: true }; + await seeder.api.appendLeaves({ treeId: TREE, leaves: leafBytes, forkId: seedForkId }); + // Sanity: the seeded state must be readable exactly as planted, or every later check is void. + const probe = await seeder.api.findLeafIndices({ + treeId: TREE, + revision: seedRevision, + leaves: [leafBytes[5]], + startIndex: 0, + }); + if (Number(probe.indices?.[0]) !== 5) { + throw new Error(`seed verification failed: ${JSON.stringify(probe)}`); + } + } finally { + await seeder.backend.destroy(); + } + }); + + afterAll(async () => { + await ws?.close(); + }); + + // Shared-fork variant used by the single-connection legs (no cross-connection ambiguity there). + const selfCheckingReads = (api: AsyncApi, tag: string): Promise[] => { + const checks: Promise[] = []; + for (let i = 0; i < N; i++) { + checks.push( + api.getTreeInfo({ treeId: TREE, revision: seedRevision }).then(r => { + if (r.treeId !== TREE) { + throw new Error(`${tag}: getTreeInfo returned treeId=${r.treeId}, expected ${TREE}`); + } + }), + ); + checks.push( + api.getStateReference({ revision: seedRevision }).then(r => { + if (!Array.isArray(r.state) || r.state.length === 0 || r.state.some(t => t.size === undefined)) { + throw new Error(`${tag}: getStateReference malformed: ${JSON.stringify(r).slice(0, 160)}`); + } + }), + ); + const idx = i; + checks.push( + api.getSiblingPath({ treeId: TREE, revision: seedRevision, leafIndex: idx }).then(r => { + if (!Array.isArray(r.path) || r.path.length === 0) { + throw new Error(`${tag}: getSiblingPath(index=${idx}) returned empty/invalid path`); + } + }), + ); + checks.push( + api.findLeafIndices({ treeId: TREE, revision: seedRevision, leaves: [leafBytes[i]], startIndex: 0 }).then(r => { + const got = r.indices?.[0]; + if (got === null || got === undefined || Number(got) !== i) { + throw new Error(`${tag}: findLeafIndices(leaf ${i}) returned index ${String(got)}, expected ${i}`); + } + }), + ); + } + return checks; + }; + + it('Leg C — single connection, sequential (sanity)', async () => { + if (!ipcPath) { + return; + } + const { api, backend } = await makeClient(1); + try { + for (let iter = 0; iter < 40; iter++) { + const idx = iter % N; + const r = await api.findLeafIndices({ + treeId: TREE, + revision: seedRevision, + leaves: [leafBytes[idx]], + startIndex: 0, + }); + const got = r.indices?.[0]; + if (got === null || got === undefined || Number(got) !== idx) { + throw new Error(`Leg C: findLeafIndices(${idx}) returned ${String(got)}`); + } + } + } finally { + await backend.destroy(); + } + }); + + it('Leg A — single connection, contended pipeline (writes + reads interleaved)', async () => { + if (!ipcPath) { + return; + } + const { api, backend } = await makeClient(1); + try { + // Private scratch fork for writes: uncommitted reads on it queue behind each write, driving + // the per-fork scheduler out of arrival order on this single connection. Reads of the seed + // fork run concurrently (different fork), stacking more reorder pressure. + const scratchForkId = (await api.createFork({ latest: true, blockNumber: 0 })).forkId; + const scratchRev = { forkId: scratchForkId, blockNumber: 0xffffffff, includeUncommitted: true }; + for (let iter = 0; iter < 25; iter++) { + const inflight: Promise[] = [...selfCheckingReads(api, `A/iter${iter}`)]; + for (let w = 0; w < 8; w++) { + const extra = [new Fr(BigInt(0x5eed_0000) + BigInt(iter * 100 + w)).toBuffer()]; + inflight.push(api.appendLeaves({ treeId: TREE, leaves: extra, forkId: scratchForkId })); + inflight.push( + api.getTreeInfo({ treeId: TREE, revision: scratchRev }).then(r => { + if (r.treeId !== TREE) { + throw new Error(`A/iter${iter}: scratch getTreeInfo returned treeId=${r.treeId}`); + } + }), + ); + } + await Promise.all(inflight); + } + } finally { + await backend.destroy(); + } + }); + + it('Leg D — multi-connection read/write soak, no churn', async () => { + if (!ipcPath) { + return; + } + const READERS = 3; + const WRITERS = 3; + const conns = await Promise.all(Array.from({ length: READERS + WRITERS }, (_, i) => makeClient(i + 1))); + try { + // Distinct identity per connection; readers keep theirs immutable so roots stay checkable. + const ids = []; + for (let c = 0; c < conns.length; c++) { + ids.push(await plantIdentity(conns[c].api, 100 + c)); + } + + const deadline = Date.now() + SOAK_MS; + const heartbeat = makeHeartbeat('D'); + let round = 0; + do { + const inflight: Promise[] = []; + for (let c = 0; c < READERS; c++) { + inflight.push( + Promise.all(identityChecks(conns[c].api, ids[c], `D/round${round}/reader${c}`, { checkRoot: true })), + ); + } + for (let c = READERS; c < conns.length; c++) { + inflight.push(writerRound(conns[c].api, ids[c], 100 + c, round, `D/round${round}/writer${c}`)); + } + // Fail fast: the first mispaired/malformed response aborts the soak with its tag. + await Promise.all(inflight); + round++; + heartbeat(round); + } while (Date.now() < deadline); + expect(round).toBeGreaterThan(0); + } finally { + await Promise.all(conns.map(({ backend }) => backend.destroy().catch(() => {}))); + } + }); + + it('Leg B — churned multi-connection workload (slot recycling)', async () => { + if (!ipcPath) { + return; + } + const CONNS = 6; + const clients = await Promise.all(Array.from({ length: CONNS }, (_, i) => makeClient(i + 1))); + const ids: ConnIdentity[] = []; + for (let c = 0; c < CONNS; c++) { + ids.push(await plantIdentity(clients[c].api, 200 + c)); + } + try { + const deadline = Date.now() + SOAK_MS; + const heartbeat = makeHeartbeat('B'); + let iter = 0; + do { + // Every connection pipelines identity-verified reads (one writer mixes in writes). Only + // the round's victim may reject, and only with its own destroy error — anything else + // (mispaired value, decode failure, another connection's socket dying) aborts the soak + // immediately with its tag. + const victim = iter % (CONNS - 1); // never the writer, so its size/root model stays valid + const perConn = clients.map(({ api }, c) => { + const tag = `B/iter${iter}/conn${c}`; + const work = + c === CONNS - 1 + ? writerRound(api, ids[c], 200 + c, iter, tag) + : Promise.all(identityChecks(api, ids[c], tag)).then(() => undefined); + return work.catch((e: Error) => { + if (c === victim && /destroyed/.test(e.message)) { + return; // the victim's own in-flight calls reject when we destroy it below + } + throw new Error(`${tag}: ${e.message}`); + }); + }); + + // Churn: while those are in flight, destroy the victim (leaving its requests in flight + // server-side) and immediately replace it (reusing the freed server slot). The + // replacement gets a fresh identity — its old fork stays behind, like a killed AVM sim's. + await clients[victim].backend.destroy().catch(() => {}); + clients[victim] = await makeClient(victim + 1); + + await Promise.all(perConn); + ids[victim] = await plantIdentity(clients[victim].api, 200 + CONNS + iter); + iter++; + heartbeat(iter); + } while (Date.now() < deadline); + } finally { + await Promise.all(clients.map(({ backend }) => backend.destroy().catch(() => {}))); + } + }); +}); diff --git a/yarn-project/world-state/src/native/wsdb_sigpipe_death.test.ts b/yarn-project/world-state/src/native/wsdb_sigpipe_death.test.ts new file mode 100644 index 000000000000..7badf2fcd980 --- /dev/null +++ b/yarn-project/world-state/src/native/wsdb_sigpipe_death.test.ts @@ -0,0 +1,86 @@ +import { Fr } from '@aztec/foundation/curves/bn254'; +import { type IpcClientAsync, UdsIpcClient } from '@aztec/ipc-runtime'; +import { MerkleTreeId } from '@aztec/stdlib/trees'; +import { AsyncApi } from '@aztec/wsdb'; + +import { jest } from '@jest/globals'; + +import { NativeWorldStateService } from './native_world_state.js'; + +// The wsdb server must survive a client that disconnects with requests still in flight: the +// reactor drops the connection's late responses (their connection's reorder state is gone, and +// client ids are never reused) and any write that races the disconnect yields EPIPE, not a +// process-killing SIGPIPE. This is the cross-process guard for +// that invariant — an in-process C++ test cannot reliably raise SIGPIPE because the kernel may +// buffer writes to a just-closed local peer. A long-lived "monitor" connection (models the TS +// world-state client, which never disconnects) proves the blast radius: it must keep reading +// correct answers while an unrelated peer churns. Mid-flight client death is routine in +// production: AVM simulator processes are killed on cancellation and teardown while their wsdb +// requests are outstanding. + +jest.setTimeout(120_000); +const TREE = MerkleTreeId.NOTE_HASH_TREE; + +describe('wsdb server survives a mid-flight client disconnect', () => { + let ws: NativeWorldStateService; + let ipcPath: string | undefined; + let seedRev: { forkId: number; blockNumber: number; includeUncommitted: boolean }; + let leafBytes: Uint8Array[]; + + const connect = async (): Promise<{ api: AsyncApi; backend: IpcClientAsync }> => { + const backend = await UdsIpcClient.connect(ipcPath!, { connectTimeoutMs: 5_000 }); + return { api: new AsyncApi(backend), backend }; + }; + + beforeAll(async () => { + ws = await NativeWorldStateService.tmp(); + try { + ipcPath = ws.getIpcPath(); + } catch { + ipcPath = undefined; // in-process build — no IPC to exercise + return; + } + // The generated AsyncApi is bytes-in/bytes-out (wire `Fr = Uint8Array`), so pass raw buffers. + leafBytes = Array.from({ length: 64 }, (_, i) => new Fr(BigInt(i) * 0x1_0000_0001n + 1n).toBuffer()); + const seeder = await connect(); + const forkId = (await seeder.api.createFork({ latest: true, blockNumber: 0 })).forkId; + seedRev = { forkId, blockNumber: 0xffffffff, includeUncommitted: true }; + await seeder.api.appendLeaves({ treeId: TREE, leaves: leafBytes, forkId }); + await seeder.backend.destroy(); + }); + + afterAll(async () => { + await ws?.close(); + }); + + it('a long-lived monitor keeps working after an unrelated client disconnects mid-flight', async () => { + if (!ipcPath) { + return; + } + const monitor = await connect(); + try { + const before = await monitor.api.getStateReference({ revision: seedRev }); + expect(Array.isArray(before.state) && before.state.length > 0).toBe(true); + + // Victim pipelines many reads, then destroys its connection WITHOUT draining them. The + // server completes them after the disconnect and must drop them cleanly. + for (let round = 0; round < 20; round++) { + const victim = await connect(); + for (let i = 0; i < 200; i++) { + void victim.api.getStateReference({ revision: seedRev }).catch(() => {}); + void victim.api.getSiblingPath({ treeId: TREE, revision: seedRev, leafIndex: i % 64 }).catch(() => {}); + } + await victim.backend.destroy(); // close mid-flight + + // The monitor must still get a correct answer; if the server died this read never + // resolves (or the connection resets), and if a stale frame leaked the decode fails. + const r = await monitor.api.getStateReference({ revision: seedRev }); + if (!Array.isArray(r.state) || r.state.length === 0) { + throw new Error(`round ${round}: monitor read malformed after churn: ${JSON.stringify(r).slice(0, 160)}`); + } + } + } finally { + await monitor.backend.destroy().catch(() => {}); + } + }); +});