diff --git a/example/calculator.cpp b/example/calculator.cpp index 5b9990ea..51d5e587 100644 --- a/example/calculator.cpp +++ b/example/calculator.cpp @@ -10,7 +10,6 @@ #include #include #include -#include #include #include #include diff --git a/example/printer.cpp b/example/printer.cpp index 2ba3f600..e3d4ce63 100644 --- a/example/printer.cpp +++ b/example/printer.cpp @@ -10,7 +10,6 @@ #include // IWYU pragma: keep #include #include -#include #include #include #include diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 49b0611a..5ad65d16 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -376,6 +376,11 @@ class EventLoop //! Hook called on the event loop thread when a client has disconnected. std::function testing_hook_disconnected; + //! Hook called at the start of ~ProxyClient, on whichever thread + //! is destroying the object, with the object being destroyed. Used by + //! tests to control timing during thread map teardown. + std::function*)> testing_hook_thread_client_destroy; + //! Miscellaneous testing hook. Called from various places with an //! argument identifying the call site (typically a string literal), so //! tests can control timing or inject behavior at specific points without @@ -431,9 +436,23 @@ struct Waiter //! to guard access to related state. Specifically, since the thread_local //! ThreadContext struct owns a Waiter, the Waiter::m_mutex is used to guard //! access to other parts of the struct to avoid needing to deal with more - //! mutexes than necessary. This mutex can be held at the same time as - //! EventLoop::m_mutex as long as Waiter::mutex is locked first and - //! EventLoop::m_mutex is locked second. + //! mutexes than necessary. + //! + //! Lock order: this mutex can be held at the same time as + //! EventLoop::m_mutex as long as Waiter::m_mutex is locked first and + //! EventLoop::m_mutex is locked second. ~ProxyServer locks them + //! in this order on the event loop thread. No code locks them in the + //! reverse order. + //! + //! Blocking rule: a thread other than the event loop thread must not hold + //! this mutex while calling EventLoop::sync() or EventLoop::post(). Those + //! calls block until the event loop thread runs the posted function, and + //! the event loop thread locks Waiter::m_mutex itself (in SetThread and + //! its disconnect callback, Waiter::post, and ~ProxyServer), so + //! holding the mutex across the call could deadlock even though the two + //! mutexes would be locked in the permitted order. This is why + //! ~ThreadContext releases the mutex before destroying ProxyClient + //! objects, whose destructor calls EventLoop::sync(). Mutex m_mutex; std::condition_variable m_cv MP_GUARDED_BY(m_mutex); std::optional> m_fn MP_GUARDED_BY(m_mutex); @@ -465,32 +484,41 @@ class Connection //! destructors of m_impl instances owned by ProxyServer objects). ~Connection() noexcept(false); - //! Register synchronous cleanup function to run on event loop thread (with - //! access to capnp thread local variables) when disconnect() is called. - //! any new i/o. + //! Register a synchronous cleanup function to run on the event loop thread + //! (with access to capnp thread-local variables) when the connection is + //! disconnected -- for either a remote disconnect (the peer closes the + //! connection) or a local one (the connection is torn down on this side). + //! Contrast onDisconnect(), whose handler runs only on a remote + //! disconnect. Returns a handle that can be passed to removeSyncCleanup() + //! to unregister the function before it runs. CleanupIt addSyncCleanup(std::function fn); void removeSyncCleanup(CleanupIt it); - //! Add disconnect handler. + //! Register a handler to run on the event loop thread when the peer + //! disconnects. The handler runs at most once, and only while this + //! Connection is still alive: if the connection is torn down locally before + //! the handler runs, the handler is not called. template void onDisconnect(F&& f) { - // Add disconnect handler to local TaskSet to ensure it is canceled and - // will never run after connection object is destroyed. But when disconnect - // handler fires, do not call the function f right away, instead add it - // to the EventLoop TaskSet to avoid "Promise callback destroyed itself" - // error in the typical case where f deletes this Connection object. - m_on_disconnect.add(m_network.onDisconnect().then( - [f = std::forward(f), this]() mutable { m_loop->m_task_set->add(kj::evalLater(kj::mv(f))); })); + // m_network.onDisconnect() fires both on a remote disconnect and on a + // local disconnect (deleting the Connection resets m_rpc_system, which + // drops capnp's last reference to the network and fulfills the + // promise). The m_alive weak_ptr tells the two apart -- it is expired + // only while the Connection is being deleted -- so f is skipped on + // local disconnects. This lets onDisconnect callbacks delete the + // Connection without a double deletion. + m_loop->m_task_set->add(m_network.onDisconnect().then( + [f = std::forward(f), alive = std::weak_ptr(m_alive)]() mutable { + if (!alive.expired()) f(); + })); } EventLoopRef m_loop; kj::Own m_stream; - LoggingErrorHandler m_error_handler{*m_loop}; - //! TaskSet used to cancel the m_network.onDisconnect() handler for remote - //! disconnections, if the connection is closed locally first by deleting - //! this Connection object. - kj::TaskSet m_on_disconnect{m_error_handler}; + //! Liveness token checked by onDisconnect() callbacks (see there). + //! Could be dropped if Connection lifetime were reference-counted (#336). + std::shared_ptr m_alive{std::make_shared()}; ::capnp::TwoPartyVatNetwork m_network; std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>> m_rpc_system; @@ -583,8 +611,8 @@ ProxyClientBase::ProxyClientBase(typename Interface::Client cli // Remove disconnect callback on cleanup so it doesn't run and try // to access this object after it's destroyed. This call needs to // run inside loop->sync() on the event loop thread because - // otherwise, if there were an ill-timed disconnect, the - // onDisconnect handler could fire and delete the Connection object + // otherwise, if there were an ill-timed disconnect, the remote + // disconnect handler could fire and delete the Connection object // before the removeSyncCleanup call. if (m_context.connection) m_context.connection->removeSyncCleanup(disconnect_cb); @@ -771,9 +799,10 @@ struct ThreadContext //! However, individual ProxyClient objects in the maps will only be //! associated with one event loop and guarded by EventLoop::m_mutex. So //! Waiter::m_mutex does not need to be held while accessing individual - //! ProxyClient instances, and may even need to be released to - //! respect lock order and avoid locking Waiter::m_mutex before - //! EventLoop::m_mutex. + //! ProxyClient instances, and must be released before destroying + //! one from a thread other than the event loop thread, because + //! ~ProxyClient calls EventLoop::sync() (see the blocking rule in + //! the Waiter::m_mutex documentation). ConnThreads callback_threads MP_GUARDED_BY(waiter->m_mutex); //! When client is making a request to a server, this is the `thread` @@ -791,6 +820,11 @@ struct ThreadContext //! to assert false if there's an attempt to execute a blocking operation //! which could deadlock the thread. bool loop_thread = false; + + //! Destructor which destroys the thread maps, coordinating with event + //! loop threads that remove entries from them concurrently when + //! connections are broken (see the code comment). + ~ThreadContext(); }; template @@ -893,10 +927,17 @@ void _Serve(EventLoop& loop, kj::Own&& stream, InitImpl& init auto it = loop.m_incoming_connections.begin(); MP_LOG(loop, Log::Info) << "IPC server: socket connected."; if (loop.testing_hook_connected) loop.testing_hook_connected(); - it->onDisconnect([&loop, it, on_disconnect = std::forward(on_disconnect)]() mutable { + // Run on_disconnect (e.g. the listener's active-connection counter + // decrement) on any disconnect. It is registered with addSyncCleanup rather + // than placed in the onDisconnect handler below because that handler + // only fires on a remote disconnect and is canceled when a connection is + // closed locally; if on_disconnect lived there, closing a connection + // locally would leave the listener's slot count stuck and stop it from + // accepting again. + it->addSyncCleanup(std::forward(on_disconnect)); + it->onDisconnect([&loop, it]() mutable { MP_LOG(loop, Log::Info) << "IPC server: socket disconnected."; loop.m_incoming_connections.erase(it); - on_disconnect(); if (loop.testing_hook_disconnected) loop.testing_hook_disconnected(); }); } diff --git a/include/mp/type-context.h b/include/mp/type-context.h index 7cfc7e79..5f7e05b2 100644 --- a/include/mp/type-context.h +++ b/include/mp/type-context.h @@ -179,10 +179,10 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& if (erase_thread) { // Look up the thread again without using existing // iterator since entry may no longer be there after - // a disconnect. Destroy node after releasing - // Waiter::m_mutex, so the ProxyClient - // destructor is able to use EventLoop::mutex - // without violating lock order. + // a disconnect. Destroy the node after releasing + // Waiter::m_mutex, so ~ProxyClient does not + // run with the mutex held (see the SetThread + // disconnect callback). ConnThreads::node_type removed; { Lock lock(thread_context.waiter->m_mutex); diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index eb2aee0c..2d694bca 100644 --- a/src/mp/proxy.cpp +++ b/src/mp/proxy.cpp @@ -111,7 +111,8 @@ Connection::~Connection() noexcept(false) // Connection destructor is always called on the event loop thread. If this // is a local disconnect, it will trigger I/O, so this needs to run on the // event loop thread, and if there was a remote disconnect, this is called - // by an onDisconnect callback directly from the event loop thread. + // by a TwoPartyVatNetwork::onDisconnect callback directly from the event + // loop thread. assert(std::this_thread::get_id() == m_loop->m_thread_id); // Try to cancel any calls that may be executing. @@ -132,13 +133,13 @@ Connection::~Connection() noexcept(false) // // Sending pending data is important if the connection is a socketpair // because when one side of the socketpair is closed, the other side doesn't - // seem to receive any onDisconnect event. So it is important for the other - // side to instead receive Cap'n Proto "release" messages (see `struct - // Release` in capnp/rpc.capnp) from local Client objects being destroyed so - // the remote side can free resources and shut down cleanly. Without this, - // when one side of a socket pair is closed the other side may not receive - // these messages, preventing the remote side from freeing ProxyServer - // resources and shutting down cleanly. + // seem to receive any TwoPartyVatNetwork::onDisconnect event. So it is + // important for the other side to instead receive Cap'n Proto "release" + // messages (see `struct Release` in capnp/rpc.capnp) from local Client + // objects being destroyed so the remote side can free resources and shut + // down cleanly. Without this, when one side of a socket pair is closed the + // other side may not receive these messages, preventing the remote side + // from freeing ProxyServer resources and shutting down cleanly. // Use kj::runCatchingExceptions instead of try/catch because on macOS with // dynamic libraries, kj::Exception typeinfo differs between libcapnp and // the calling binary, so catch (const kj::Exception&) silently fails to @@ -184,17 +185,18 @@ Connection::~Connection() noexcept(false) // connection implementing the Init interface and handling the Init.makeX() calls. // // Either way when a connection is closed, capnp behavior is to call all - // ProxyServer object destructors first, and then trigger an onDisconnect - // callback. + // ProxyServer object destructors first, and then trigger a + // TwoPartyVatNetwork::onDisconnect callback. // - // On incoming side of the connection, the onDisconnect callback is written - // to delete the Connection object from the m_incoming_connections and call - // this destructor which calls Connection::disconnect. + // On incoming side of the connection, the TwoPartyVatNetwork::onDisconnect + // callback is written to delete the Connection object from the + // m_incoming_connections list and call this destructor. // - // On the outgoing side, the Connection object is owned by top level client - // object client, which onDisconnect handler doesn't have ready access to, - // so onDisconnect handler just calls Connection::disconnect directly - // instead. + // On the outgoing side, the Connection object is heap-allocated and owned + // by a top-level ProxyClient object. In this case, the + // TwoPartyVatNetwork::onDisconnect handler deletes the Connection object + // directly, calling this destructor, and loop below sets Connection + // pointers in all associated ProxyClient objects to null. // // Either way disconnect code runs in the event loop thread and called both // on clean and unclean shutdowns. In unclean shutdown case when the @@ -218,7 +220,8 @@ CleanupIt Connection::addSyncCleanup(std::function fn) // order should not be significant because the cleanup callbacks run // synchronously in a single batch when the connection is broken, and they // only reset the connection pointers in the client objects without actually - // deleting the client objects. + // deleting the client objects, or update ListenConnections max_connections + // bookkeeping. return m_sync_cleanup_fns.emplace(m_sync_cleanup_fns.begin(), std::move(fn)); } @@ -394,20 +397,26 @@ std::tuple SetThread(GuardedRef threads, Connecti } if (inserted) { thread->second.emplace(make_thread(), connection, /* destroy_connection= */ false); - thread->second->m_disconnect_cb = connection->addSyncCleanup([threads, thread] { - // Note: it is safe to use the `thread` iterator in this cleanup - // function, because the iterator would only be invalid if the map entry - // was removed, and if the map entry is removed the ProxyClient - // destructor unregisters the cleanup. - - // Connection is being destroyed before thread client is, so reset - // thread client m_disconnect_cb member so thread client destructor does not - // try to unregister this callback after connection is destroyed. - thread->second->m_disconnect_cb.reset(); - - // Remove connection pointer about to be destroyed from the map - const Lock lock(threads.mutex); - threads.ref.erase(thread); + thread->second->m_disconnect_cb = connection->addSyncCleanup([threads, connection] { + // Remove and destroy this connection's map entry, unless the + // thread owning the map is exiting and ~ThreadContext already took + // the entry, in which case that thread destroys it. Look the entry + // up by key rather than capturing the iterator, which would be + // invalid in that case. + ConnThreads::node_type removed; + { + const Lock lock(threads.mutex); + auto it = threads.ref.find(connection); + if (it == threads.ref.end()) return; + // Reset so ~ProxyClient does not try to unregister + // this callback, which is already running. + it->second->m_disconnect_cb.reset(); + removed = threads.ref.extract(it); + } + // The entry is destroyed here with Waiter::m_mutex released, as in + // ~ThreadContext. (Holding it would be safe on the event loop + // thread, where EventLoop::sync() runs directly, but releasing it + // keeps the two consistent.) }); } return {thread, inserted}; @@ -415,6 +424,9 @@ std::tuple SetThread(GuardedRef threads, Connecti ProxyClient::~ProxyClient() { + EventLoop& loop{*m_context.loop}; + if (loop.testing_hook_thread_client_destroy) loop.testing_hook_thread_client_destroy(this); + // If thread is being destroyed before connection is destroyed, remove the // cleanup callback that was registered to handle the connection being // destroyed before the thread being destroyed. @@ -424,13 +436,42 @@ ProxyClient::~ProxyClient() // between this thread trying to remove the callback and the disconnect // handler attempting to call it. m_context.loop->sync([&]() { - if (m_disconnect_cb) { + // Skip if the connection was destroyed while this thread waited + // for the event loop: ~Connection has already run and freed the + // cleanup list m_disconnect_cb points into, and the ProxyClientBase + // disconnect callback has nulled m_context.connection. (The + // SetThread callback resets m_disconnect_cb only when it finds + // this object in the thread map, so if ~ThreadContext took the + // entry first, m_disconnect_cb is still set here.) + if (m_disconnect_cb && m_context.connection) { m_context.connection->removeSyncCleanup(*m_disconnect_cb); } }); } } +ThreadContext::~ThreadContext() +{ + // Server threads created by ProxyServer::makeThread have no + // waiter here: ~ProxyServer moves it away and clears the maps + // before the thread exits. + if (!waiter) return; + + // Take the thread client maps under Waiter::m_mutex, because the SetThread + // disconnect callback removes entries from them concurrently on the event + // loop thread when connections are broken, and whichever side removes an + // entry destroys it. Destroy the maps with the mutex released, because + // ~ProxyClient blocks in EventLoop::sync() and the event loop + // thread locks Waiter::m_mutex itself (see the blocking rule in the + // Waiter::m_mutex documentation). + ConnThreads request, callback; + { + const Lock lock(waiter->m_mutex); + request.swap(request_threads); + callback.swap(callback_threads); + } +} + ProxyServer::ProxyServer(Connection& connection, ThreadContext& thread_context, std::thread&& thread) : m_loop{*connection.m_loop}, m_thread_context(thread_context), m_thread(std::move(thread)) { diff --git a/test/mp/test/connect_tests.cpp b/test/mp/test/connect_tests.cpp index 4e4f051f..7f522ad6 100644 --- a/test/mp/test/connect_tests.cpp +++ b/test/mp/test/connect_tests.cpp @@ -118,11 +118,11 @@ KJ_TEST("ConnectStream defers disconnect failure to the first IPC request for in } catch (const std::runtime_error& e) { std::string_view reason = e.what(); - // There is a race between the event loop detecting the disconnect - // and foo->add() being called. If the onDisconnect callback fires - // first and nulls m_context.connection, the error is "called after - // disconnect"; if foo->add() submits before the callback fires, - // the error is "interrupted by disconnect". + // There is a race between the event loop detecting the disconnect and + // foo->add() being called. If the TwoPartyVatNetwork::onDisconnect + // callback fires first and nulls m_context.connection, the error is + // "called after disconnect"; if foo->add() submits before the callback + // fires, the error is "interrupted by disconnect". KJ_EXPECT(reason == "IPC client method called after disconnect." || reason == "IPC client method call interrupted by disconnect."); } diff --git a/test/mp/test/listen_tests.cpp b/test/mp/test/listen_tests.cpp index 8d06cb35..e7c98e6b 100644 --- a/test/mp/test/listen_tests.cpp +++ b/test/mp/test/listen_tests.cpp @@ -18,6 +18,7 @@ #include #include #include +#include #include #include #include @@ -192,6 +193,34 @@ KJ_TEST("ListenConnections enforces a local connection limit") KJ_EXPECT(client3->client->add(3, 4) == 7); } +KJ_TEST("ListenConnections resumes after a local disconnect") +{ + // A connection closed locally (here by erasing it from the incoming- + // connection list) must still free the listener's slot so the listener + // resumes accepting. The counter decrement runs as an addSyncCleanup cleanup; + // if it were in the onDisconnect handler it would be canceled by the + // local close and the slot would stay stuck. + ListenSetup server(/*max_connections=*/1); + + auto client1 = std::make_unique(server.listener.MakeConnectedSocket()); + server.WaitForConnectedCount(1); + KJ_EXPECT(client1->client->add(1, 2) == 3); + + auto client2 = std::make_unique(server.listener.MakeConnectedSocket()); + (**server.m_loop_ref).sync([] {}); + KJ_EXPECT(server.ConnectedCount() == 1); + + // Close the first connection locally on the event loop thread. + EventLoop& loop{**server.m_loop_ref}; + loop.sync([&] { + KJ_REQUIRE(loop.m_incoming_connections.size() == 1); + loop.m_incoming_connections.pop_front(); + }); + + server.WaitForConnectedCount(2); + KJ_EXPECT(client2->client->add(2, 3) == 5); +} + KJ_TEST("ListenConnections accepts multiple connections") { // With max-connections=2, two clients should be accepted and usable at the diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 5ecb7cc4..ca4edcfc 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -419,7 +420,8 @@ KJ_TEST("Calling async IPC method with a remote disconnect while results are bui // setting request_canceled) and the worker would throw InterruptException // instead of proceeding into getResults(). Keeping it alive matches the // window in the original report, where the worker races with capnp's own - // internal teardown, which runs before any onDisconnect notification. + // internal teardown, which runs before any TwoPartyVatNetwork::onDisconnect + // notification. TestSetup setup{/*client_owns_connection=*/false}; ProxyClient* foo = setup.client.get(); @@ -498,6 +500,54 @@ KJ_TEST("Worker thread destroyed before it is initialized") EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); } +KJ_TEST("Thread exiting while its connection is destroyed") +{ + // Regression test for a race between a thread exiting after making IPC + // calls and its connection being destroyed on the event loop thread. + // ~ThreadContext on the exiting thread and the SetThread disconnect + // callback run by ~Connection both remove the thread's map entries for + // the connection, and previously nothing synchronized them, so both could + // destroy the same ProxyClient object. + // + // The testing_hook_thread_client_destroy hook, called at the start of + // ~ProxyClient, blocks the exiting thread inside its first map + // entry destructor while the main thread destroys the connection. The + // disconnect callback must leave that entry alone: it resets + // m_disconnect_cb only when it finds the entry in the map and takes over + // destroying it, so the entry's m_disconnect_cb must still be set when + // the exiting thread resumes. + TestSetup setup{/*client_owns_connection=*/false}; + ProxyClient* foo = setup.client.get(); + foo->initThreadMap(); + setup.server->m_impl->m_fn = [] {}; + EventLoop& loop = *foo->m_context.loop; + + std::promise caller_exiting, release_caller; + bool caller_blocked{false}; // caller thread only + bool disconnect_cb_set{false}; // caller thread, read after join + loop.testing_hook_thread_client_destroy = [&](ProxyClient* client) { + // The hook also runs on the event loop thread for entries the + // disconnect callback destroys. Block only the exiting caller thread, + // in the first destructor it runs. + if (std::this_thread::get_id() == loop.m_thread_id || caller_blocked) return; + caller_blocked = true; + caller_exiting.set_value(); + release_caller.get_future().get(); + disconnect_cb_set = client->m_disconnect_cb.has_value(); + }; + + // Make a call taking an mp.Context argument, which adds callback and + // request thread entries for the connection to the caller's thread-local + // ThreadContext maps. They are destroyed when the thread exits. + std::thread caller{[&] { foo->callFnAsync(); }}; + caller_exiting.get_future().get(); + setup.client_disconnect(); + release_caller.set_value(); + caller.join(); + loop.testing_hook_thread_client_destroy = nullptr; + KJ_EXPECT(disconnect_cb_set); +} + KJ_TEST("Calling async IPC method, with server disconnect racing the call") { // Regression test for bitcoin/bitcoin#34777 heap-use-after-free where @@ -734,6 +784,85 @@ KJ_TEST("Async cleanup thread has OS thread name") } #endif // HAVE_PTHREAD_GETNAME_NP +KJ_TEST("onDisconnect handler does not run after the connection is destroyed") +{ + // Regression test for a race condition where an onDisconnect handler + // could fire even after the connection had already been disconnected + // locally. Local disconnects are supposed to preempt onDisconnect + // handlers so they are able to free Connection objects without risking double + // deletions. + // + // To hit the previous race window deterministically without relying on + // socket timing, this test exploits two ordering facts: + // m_network.onDisconnect() hands out branches of a forked promise that fire + // in the order they were added, and kj::evalLater tasks queued during one + // event loop turn run on the next turn in FIFO order. A "destroyer" branch + // is registered before the handler under test, so when the peer disconnects + // the destroyer fires first and queues an evalLater task that destroys the + // Connection ahead of the handler. The handler records whether it ran after + // the Connection was destroyed, reading only test-owned flags (never the + // freed Connection), so a violation is a deterministic wrong result rather + // than a flaky crash. + // + // The Connection must be destroyed from that separate task, not from inside + // an onDisconnect continuation: destroying it there tears down m_network + // while a continuation of m_network's promise is still running, and crashes. + + std::atomic connection_destroyed{false}; + std::atomic handler_ran{false}; + std::atomic handler_ran_after_destroy{false}; + + std::thread thread{[&] { + EventLoop loop("mptest", [](mp::LogMessage log) { + KJ_LOG(INFO, log.level, log.message); + if (log.level == mp::Log::Raise) throw std::runtime_error(log.message); + }); + auto pipe = loop.m_io_context.provider->newTwoWayPipe(); + + // Server-side connection whose onDisconnect handler is under test. + auto server_conn = std::make_unique( + loop, kj::mv(pipe.ends[0]), [&](Connection& connection) { + return ::capnp::Capability::Client(kj::heap>( + std::make_shared(), connection)); + }); + + // Raw peer end of the pipe. Closing it makes the server connection's + // m_network.onDisconnect() resolve, i.e. a real remote disconnect. + kj::Own peer_end = kj::mv(pipe.ends[1]); + + // Destroyer branch, registered BEFORE the handler under test so it + // fires first and its destroy task E_d is queued before the buggy + // bounced handler task E_f. It schedules the destroy on a *separate* + // task, so the connection is not torn down from inside an onDisconnect + // continuation. + loop.m_task_set->add(server_conn->m_network.onDisconnect().then([&] { + loop.m_task_set->add(kj::evalLater([&] { + server_conn.reset(); + connection_destroyed = true; + })); + })); + + // Handler under test. It intentionally does not dereference the + // Connection; it only records whether it ran after the connection was + // destroyed, so the result is deterministic. + server_conn->onDisconnect([&] { + handler_ran = true; + handler_ran_after_destroy = connection_destroyed.load(); + }); + + // Trigger the remote disconnect once the loop is running. + loop.m_task_set->add(kj::evalLater([&] { peer_end = nullptr; })); + + loop.loop(); + }}; + thread.join(); + + // The handler must have run (both versions schedule it), but it must never + // run after the connection was destroyed. + KJ_EXPECT(handler_ran); + KJ_EXPECT(!handler_ran_after_destroy); +} + KJ_TEST("Call async IPC method without thread or pool errors correctly") { TestSetup setup;