From 801380139be1f0ce8574158fc17666fc39c8d02e Mon Sep 17 00:00:00 2001 From: Ryan Ofsky Date: Thu, 3 Sep 2026 15:23:49 -0400 Subject: [PATCH 1/8] This is a documentation-only change meant to make upcoming commits easier to understand. Co-Authored-By: Claude Opus 4.8 (1M context) --- include/mp/proxy-io.h | 28 +++++++++++++++----------- src/mp/proxy.cpp | 36 ++++++++++++++++++---------------- test/mp/test/connect_tests.cpp | 10 +++++----- test/mp/test/test.cpp | 3 ++- 4 files changed, 43 insertions(+), 34 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 49b0611a..cdaf8665 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -465,18 +465,24 @@ 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. + //! Add a remote disconnect handler, run when the peer closes the + //! connection. The handler is canceled if the connection is disconnected + //! locally first (which destroys m_on_disconnect). 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 + // Add the handler to the local TaskSet to ensure it is canceled and + // will never run after the connection object is destroyed. But when the // 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. @@ -487,9 +493,9 @@ class Connection 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. + //! TaskSet holding the m_network.onDisconnect() handler for remote + //! disconnections. Reset to cancel the handler if the connection is closed + //! locally first by deleting this Connection object. kj::TaskSet m_on_disconnect{m_error_handler}; ::capnp::TwoPartyVatNetwork m_network; std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>> m_rpc_system; @@ -583,8 +589,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); diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index eb2aee0c..b49180ef 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 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/test.cpp b/test/mp/test/test.cpp index 5ecb7cc4..4b9d3412 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -419,7 +419,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(); From e9bbe34e30d3f1fd8a1cbc0d74e124570ec59fce Mon Sep 17 00:00:00 2001 From: Ryan Ofsky Date: Thu, 10 Sep 2026 16:09:00 -0400 Subject: [PATCH 2/8] Correct the ThreadContext "Synchronization note", which said Waiter::m_mutex must not be locked before EventLoop::m_mutex. That is the reverse of the documented and actual lock order (Waiter::m_mutex first, as ~ProxyServer does). The constraint it was reaching for is the EventLoop blocking rule now documented on Waiter::m_mutex. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01YKQfSnMnUzqyDxKp7GpFam --- include/mp/proxy-io.h | 27 +++++++++++++++++++++------ include/mp/type-context.h | 8 ++++---- 2 files changed, 25 insertions(+), 10 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index cdaf8665..7cacdea1 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -431,9 +431,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); @@ -777,9 +791,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` 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); From 33bd6f833f5af865977691c6176923d025d82d71 Mon Sep 17 00:00:00 2001 From: Ryan Ofsky Date: Thu, 3 Sep 2026 15:30:21 -0400 Subject: [PATCH 3/8] proxy-io: fix listener stuck at capacity after a local disconnect Currently, a ListenConnections listener that reaches its max-connection limit stops accepting new connections permanently if one of its connections is closed locally instead of by a remote disconnect. Closing a connection locally (e.g. erasing it from m_incoming_connections) leaves the listener's active-connection count stuck at the limit, so it never resumes accepting. This happens because the count is decremented by a callback which only fires on a remote disconnects, not local disconnects. Fix by moving the decrement to callback which fires on both local and remote disconnects. Add a regression test that closes a connection locally and checks the listener resumes accepting; it fails before this change (the listener never accepts the waiting client) and passes after. Co-Authored-By: Enoch Azariah Co-Authored-By: Claude Opus 4.8 (1M context) --- include/mp/proxy-io.h | 11 +++++++++-- src/mp/proxy.cpp | 3 ++- test/mp/test/listen_tests.cpp | 28 ++++++++++++++++++++++++++++ 3 files changed, 39 insertions(+), 3 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 7cacdea1..69033b3e 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -914,10 +914,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/src/mp/proxy.cpp b/src/mp/proxy.cpp index b49180ef..98429d1b 100644 --- a/src/mp/proxy.cpp +++ b/src/mp/proxy.cpp @@ -220,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)); } diff --git a/test/mp/test/listen_tests.cpp b/test/mp/test/listen_tests.cpp index 8d06cb35..402d3c10 100644 --- a/test/mp/test/listen_tests.cpp +++ b/test/mp/test/listen_tests.cpp @@ -192,6 +192,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 From 7cbade8b18a808ce4cf0f60520026fd97946b97f Mon Sep 17 00:00:00 2001 From: Ryan Ofsky Date: Thu, 3 Sep 2026 16:34:56 -0400 Subject: [PATCH 4/8] proxy-io: fix race deleting a disconnected Connection twice Fix a use-after-free, possible since the destroy_connection option was added in 2019 (c685fa9): a Connection's disconnect handler could run after the Connection had already been destroyed, deleting it a second time and crashing. Reported by enirox001 in https://github.com/bitcoin-core/libmultiprocess/pull/335#discussion_r3821831654 Give each Connection a shared_ptr "alive" token that disconnect handlers hold a weak_ptr to and check before running, so a handler is skipped once its Connection is gone. Having this check also enables the simplifications described below. Previously each Connection kept its disconnect handlers in its own kj::TaskSet, and when the network disconnected it moved a handler onto the shared event loop TaskSet with kj::evalLater. Destroying the Connection destroyed that per-connection TaskSet, canceling a still-pending handler -- but a handler already moved onto the shared TaskSet was no longer canceled and could run after the Connection was gone. (The evalLater step existed only to avoid a "promise callback destroyed itself" error when a handler deletes its own Connection, which the per-connection TaskSet made possible.) With the token doing the cancellation, neither the per-connection TaskSet nor the evalLater step is needed, and both are removed. Co-Authored-By: Enoch Azariah Co-Authored-By: Claude Fable 5 --- example/calculator.cpp | 1 - example/printer.cpp | 1 - include/mp/proxy-io.h | 33 ++++++++------- test/mp/test/listen_tests.cpp | 1 + test/mp/test/test.cpp | 80 +++++++++++++++++++++++++++++++++++ 5 files changed, 99 insertions(+), 17 deletions(-) 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 69033b3e..85117068 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -489,28 +489,31 @@ class Connection CleanupIt addSyncCleanup(std::function fn); void removeSyncCleanup(CleanupIt it); - //! Add a remote disconnect handler, run when the peer closes the - //! connection. The handler is canceled if the connection is disconnected - //! locally first (which destroys m_on_disconnect). + //! 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 the handler to the local TaskSet to ensure it is canceled and - // will never run after the connection object is destroyed. But when the - // 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 holding the m_network.onDisconnect() handler for remote - //! disconnections. Reset to cancel the handler 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; diff --git a/test/mp/test/listen_tests.cpp b/test/mp/test/listen_tests.cpp index 402d3c10..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 diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 4b9d3412..56a84dda 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -735,6 +736,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; From 3a4a5eb5f54e8a0f75a5ff7b83dea2441baf87e4 Mon Sep 17 00:00:00 2001 From: Ryan Ofsky Date: Fri, 11 Sep 2026 10:22:28 -0400 Subject: [PATCH 5/8] Fix a race between a thread exiting after making IPC calls and its connection being destroyed on the event loop thread, which could destroy the same ProxyClient object twice. ~ThreadContext destroyed the thread-local request_threads/callback_threads maps with no locking while the SetThread cleanup callback run by ~Connection erased entries from the same maps. When both ran at once, each side destroyed the entry's ProxyClient, and ~Connection then ran the ProxyClientBase disconnect callback on the freed map node (heap-use-after-free, then a glibc "double free or corruption" abort). Fix by making map entry removal decide which side destroys an entry: ~ThreadContext and the SetThread callback each remove entries under Waiter::m_mutex before destroying them, and a side that finds an entry already gone leaves it to the other. See the code comments for why the entries are destroyed with the mutex released. Add a regression test, "Thread exiting while its connection is destroyed", which uses a new testing_hook_thread_client_destroy hook to interleave the two sides deterministically and fails on every run without the fix. The race is long-standing and reachable on master via connections created by ConnectStream, whose onDisconnect handler deletes the client Connection on the event loop thread when the peer disconnects. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01BnBLP1xuPf4fLpnQto8mEX Claude-Session: https://claude.ai/code/session_01YKQfSnMnUzqyDxKp7GpFam --- include/mp/proxy-io.h | 10 +++++++ src/mp/proxy.cpp | 68 +++++++++++++++++++++++++++++++++---------- test/mp/test/test.cpp | 48 ++++++++++++++++++++++++++++++ 3 files changed, 111 insertions(+), 15 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 85117068..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 @@ -815,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 diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index 98429d1b..2d694bca 100644 --- a/src/mp/proxy.cpp +++ b/src/mp/proxy.cpp @@ -397,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}; @@ -418,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. @@ -427,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/test.cpp b/test/mp/test/test.cpp index 56a84dda..ca4edcfc 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -500,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 From d746bb39fa74c690e31d88a7c09a881733743654 Mon Sep 17 00:00:00 2001 From: Ryan Ofsky Date: Fri, 31 Jul 2026 15:19:30 -0400 Subject: [PATCH 6/8] proxy-io: add Connection::disconnect() separating teardown from destruction Split connection teardown out of ~Connection into an idempotent disconnect() method, with the destructor delegating to it. For existing callers, this is a behavior-neutral refactor: the same steps run in the same order on destruction. Having a separate disconnect() method allows severing a connection while keeping the Connection object alive, which the next commits use to let shutdown code wait for in-flight server call bodies to finish after a disconnect (bitcoin/bitcoin#35845). Two details are new in the disconnect() method which were not present in the destructor method: - disconnect() expires the m_alive token explicitly, where previously it was expired implicitly by member destruction. This keeps onRemoteDisconnect able to distinguish a local disconnect from a remote one when a connection is severed without destroying the object (see the disconnect() code comment). - disconnect() explicitly releases m_thread_pool and m_thread_map so worker thread teardown happens at disconnect time whether or not the object is destroyed right away. Previously this happened implicitly during member destruction. Co-Authored-By: Claude Fable 5 --- include/mp/proxy-io.h | 56 +++++++++++++++++++++++------------ src/mp/proxy.cpp | 68 +++++++++++++++++++++++++++++++++++-------- test/mp/test/test.cpp | 2 +- 3 files changed, 94 insertions(+), 32 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 5ad65d16..68c19883 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -463,27 +463,42 @@ struct Waiter //! on the event loop thread. //! In addition to Cap'n Proto state, it also holds lists of callbacks to run //! when the connection is closed. +//! +//! A Connection may be severed with disconnect() and then kept alive (rather +//! than destroyed) so callers can wait for in-flight server calls to finish. +//! Once disconnect() has run the object holds no transport, RPC system, or +//! worker threads, so the only valid operations on it are destruction and +//! calling disconnect() again (a no-op). Methods that perform I/O or make +//! calls must not be used after disconnect(). class Connection { public: Connection(EventLoop& loop, kj::Own&& stream_) : m_loop(loop), m_stream(kj::mv(stream_)), - m_network(*m_stream, ::capnp::rpc::twoparty::Side::CLIENT, ::capnp::ReaderOptions()), - m_rpc_system(::capnp::makeRpcClient(m_network)) {} + m_network(std::in_place, *m_stream, ::capnp::rpc::twoparty::Side::CLIENT, ::capnp::ReaderOptions()), + m_rpc_system(::capnp::makeRpcClient(*m_network)) {} Connection(EventLoop& loop, kj::Own&& stream_, const std::function<::capnp::Capability::Client(Connection&)>& make_client) : m_loop(loop), m_stream(kj::mv(stream_)), - m_network(*m_stream, ::capnp::rpc::twoparty::Side::SERVER, ::capnp::ReaderOptions()), - m_rpc_system(::capnp::makeRpcServer(m_network, make_client(*this))) {} - - //! Run cleanup functions. Must be called from the event loop thread. First - //! calls synchronous cleanup functions while blocked (to free capnp - //! Capability::Client handles owned by ProxyClient objects), then schedules - //! asynchronous cleanup functions to run in a worker thread (to run - //! destructors of m_impl instances owned by ProxyServer objects). + m_network(std::in_place, *m_stream, ::capnp::rpc::twoparty::Side::SERVER, ::capnp::ReaderOptions()), + m_rpc_system(::capnp::makeRpcServer(*m_network, make_client(*this))) {} + + //! Destroy the connection. Calls disconnect() if it has not been called + //! already. Must be called from the event loop thread. ~Connection() noexcept(false); + //! Sever the connection without destroying this object: close the transport + //! so the peer observes the disconnect, and run the connection's cleanup + //! handlers. Idempotent, and called automatically by the destructor -- + //! calling it directly is only needed to disconnect while keeping the object + //! alive (e.g. to wait for in-flight calls to drain afterward). Must be + //! called from the event loop thread. + //! + //! Cancels the KJ promise of any in-flight call, but a server method body + //! already dispatched to a worker thread runs to completion. + void disconnect(); + //! 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 @@ -501,14 +516,13 @@ class Connection template void onDisconnect(F&& 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( + // m_network->onDisconnect() fires both on a remote disconnect and on a + // local one (disconnecting 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 -- disconnect() expires it, 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(); })); @@ -519,7 +533,11 @@ class Connection //! 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; + //! Wrapped in std::optional so disconnect() can tear it down (along with + //! the stream) to sever the transport while this object stays alive. + //! Closing the stream is what makes the peer observe the disconnect: it + //! reads EOF and fails its outstanding calls with DISCONNECTED errors. + std::optional<::capnp::TwoPartyVatNetwork> m_network; std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>> m_rpc_system; // ThreadMap interface client, used to create a remote server thread when an diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index 2d694bca..e59d7693 100644 --- a/src/mp/proxy.cpp +++ b/src/mp/proxy.cpp @@ -13,6 +13,7 @@ #include #include #include // IWYU pragma: keep +#include #include #include #include @@ -114,6 +115,28 @@ Connection::~Connection() noexcept(false) // by a TwoPartyVatNetwork::onDisconnect callback directly from the event // loop thread. assert(std::this_thread::get_id() == m_loop->m_thread_id); + disconnect(); +} + +void Connection::disconnect() +{ + // Disconnecting triggers I/O and tears down capnp state, so it must run on + // the event loop thread, like the destructor. + assert(std::this_thread::get_id() == m_loop->m_thread_id); + + // m_network is reset at the end of teardown below, so treat it being null + // as the "already disconnected" state: a second call (including the one + // from the destructor) is a no-op. + if (!m_network) return; + + // Expire m_alive before severing the connection below so onDisconnect + // handlers will not trigger and delete this Connection object. The + // onDisconnect handlers trigger on remote disconnects and automatically + // delete Connection objects. But on local disconnects, they should not + // trigger, because local code that disconnects is responsible for freeing + // Connection objects, and it may want to wait for in-flight calls to finish + // before destroying them. + m_alive.reset(); // Try to cancel any calls that may be executing. m_canceler.cancel("Interrupted by disconnect"); @@ -202,12 +225,32 @@ Connection::~Connection() noexcept(false) // on clean and unclean shutdowns. In unclean shutdown case when the // connection is broken, sync and async cleanup lists will be filled with // callbacks. In the clean shutdown case both lists will be empty. - Lock lock{m_loop->m_mutex}; - while (!m_sync_cleanup_fns.empty()) { - CleanupList fn; - fn.splice(fn.begin(), m_sync_cleanup_fns, m_sync_cleanup_fns.begin()); - Unlock(lock, fn.front()); + { + Lock lock{m_loop->m_mutex}; + while (!m_sync_cleanup_fns.empty()) { + CleanupList fn; + fn.splice(fn.begin(), m_sync_cleanup_fns, m_sync_cleanup_fns.begin()); + Unlock(lock, fn.front()); + } } + + // Release Thread capabilities so idle worker threads are stopped and + // joined at disconnect time, whether or not this object is destroyed right + // away. (A worker thread currently executing a call body is unaffected: + // its ProxyServer object is pinned by the post() call and released + // when the body finishes.) + m_thread_pool.clear(); + m_thread_map = nullptr; + + // Destroy the network and close the stream so the peer observes the + // disconnect, reading EOF and failing its outstanding calls with + // DISCONNECTED errors. This has to be explicit because when disconnect() + // is called without destroying this object, nothing else severs the + // transport: m_rpc_system.reset() above stops reading from the stream but + // does not reliably close it. The network is destroyed first since it + // references the stream. + m_network.reset(); + m_stream = nullptr; } CleanupIt Connection::addSyncCleanup(std::function fn) @@ -436,13 +479,14 @@ ProxyClient::~ProxyClient() // between this thread trying to remove the callback and the disconnect // handler attempting to call it. m_context.loop->sync([&]() { - // 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.) + // Skip if the connection was disconnected while this thread waited + // for the event loop: Connection::disconnect() 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); } diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index ca4edcfc..8bab6a06 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -835,7 +835,7 @@ KJ_TEST("onDisconnect handler does not run after the connection is destroyed") // 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(server_conn->m_network->onDisconnect().then([&] { loop.m_task_set->add(kj::evalLater([&] { server_conn.reset(); connection_destroyed = true; From 98d28df63e8ef009e24a2b3bb67040c253b30f2c Mon Sep 17 00:00:00 2001 From: Ryan Ofsky Date: Fri, 31 Jul 2026 15:21:48 -0400 Subject: [PATCH 7/8] proxy-io: add Connection::waitDrained() to wait for in-flight server calls Add a per-connection ServerObjectTracker counting live ProxyServer objects, incremented in the ProxyServerBase constructor and decremented in its destructor, with Connection::waitDrained() blocking until the count reaches zero and Connection::pendingServerObjects() exposing it for logging. Disconnecting a connection cancels the KJ promise of an in-flight call, but a C++ server method body already dispatched to a worker thread runs to completion. Counting live server objects turns Cap'n Proto's object lifetime rules into a usable quiescence signal: a ProxyServer object is not destroyed until its outstanding calls finish (the target capability is kept alive for the duration of a call and pinned by post()/PassField via thisCap()), so after disconnect() the count drains to zero exactly when no server call body is still executing. Waiting for that lets shutdown code avoid freeing application state that a still-running call body dereferences (bitcoin/bitcoin#35845). The tracker is held via shared_ptr by the Connection and by every ProxyServer object because objects kept alive by in-flight calls can outlive the Connection on some teardown paths (see ~ProxyServerBase), and their destructors must decrement state that is still valid. It must be declared before m_rpc_system, whose construction creates the bootstrap server object that registers itself with the tracker. Co-Authored-By: Claude Fable 5 --- include/mp/proxy-io.h | 110 +++++++++++++++++++++++++++++++++++++++++- include/mp/proxy.h | 7 +++ 2 files changed, 116 insertions(+), 1 deletion(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 68c19883..54253c18 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -458,6 +458,81 @@ struct Waiter std::optional> m_fn MP_GUARDED_BY(m_mutex); }; +//! Counter tracking the number of live ProxyServer objects associated with a +//! Connection, used to wait for a disconnected connection's server side to +//! become quiescent. +//! +//! This count is non-zero as long as the client is holding any object handles +//! or as long as any server methods are running, even after the client as +//! disconnected or freed all its handles. When this is zero it means no more +//! ProxyServer method calls can originate from the connection and it is safe to +//! free any resources accessed through the connection. +// +//! Why counting live server objects is a valid "no server call body running" +//! signal: a ProxyServer object is reference counted and is not destroyed +//! until its outstanding calls finish. Cap'n Proto keeps the target capability +//! alive for the duration of a call, and the mp.Context PassField overload and +//! ProxyServer::post() additionally pin it (self = thisCap()) until +//! the call body running on a worker thread completes and its result is +//! delivered. So "object destroyed" implies "its call bodies finished", and a +//! connection whose live-object count reached zero after a disconnect has no +//! server code running. This matters because disconnecting only cancels the +//! KJ promise of an in-flight call; it does not interrupt a call body that +//! was already dispatched to a worker thread (see Connection::disconnect). +//! +//! The counter is held via shared_ptr by the Connection and by every +//! ProxyServer object created for the connection, because a ProxyServer +//! object kept alive by an in-flight call can outlive the Connection (see +//! ~ProxyServerBase), and its destructor must decrement state that is still +//! valid. +//! +//! ProxyServer and ProxyServer are separate +//! specializations (not ProxyServerBase instances) and are intentionally not +//! counted: every application method body runs on an interface ProxyServer, +//! which is counted and stays alive for the duration of the body, so counting +//! those is sufficient. +struct ServerObjectTracker +{ + //! Called from the ProxyServerBase constructor (on the event loop thread). + void addServerObject() + { + const Lock lock(m_mutex); + m_count += 1; + } + + //! Called from the ProxyServerBase destructor (on the event loop thread). + void removeServerObject() + { + { + const Lock lock(m_mutex); + assert(m_count > 0); + m_count -= 1; + } + m_cv.notify_all(); + } + + //! Return the current count. May be called from any thread. + size_t pendingServerObjects() const + { + const Lock lock(m_mutex); + return m_count; + } + + //! Block until no server objects remain. Must NOT be called from the event + //! loop thread: in-flight call bodies need the event loop to deliver their + //! results before their server objects are destroyed, so blocking the loop + //! here would deadlock. + void waitDrained() + { + Lock lock(m_mutex); + m_cv.wait(lock.m_lock, [this]() MP_REQUIRES(m_mutex) { return m_count == 0; }); + } + + mutable Mutex m_mutex; + std::condition_variable m_cv; + size_t m_count MP_GUARDED_BY(m_mutex){0}; +}; + //! Object holding network & rpc state associated with either an incoming server //! connection, or an outgoing client connection. It must be created and destroyed //! on the event loop thread. @@ -499,6 +574,12 @@ class Connection //! already dispatched to a worker thread runs to completion. void disconnect(); + //! Return server object tracker allowing clients to block until no + //! ProxyServer objects associated with this connection remain, i.e. until + //! no server call body is still executing (see ServerObjectTracker). + using Tracker = std::shared_ptr; + Tracker tracker() { return m_server_objects; } + //! 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 @@ -538,6 +619,18 @@ class Connection //! Closing the stream is what makes the peer observe the disconnect: it //! reads EOF and fails its outstanding calls with DISCONNECTED errors. std::optional<::capnp::TwoPartyVatNetwork> m_network; + + //! Tracker for live ProxyServer objects associated with this connection, + //! used by waitDrained(). Held via shared_ptr because ProxyServer objects + //! kept alive by in-flight calls can outlive the Connection (see + //! ServerObjectTracker and ~ProxyServerBase). + //! + //! Must be declared before m_rpc_system: constructing m_rpc_system runs + //! the make_client callback, which creates the bootstrap (Init) server + //! object, whose ProxyServerBase constructor registers itself with this + //! tracker. + std::shared_ptr m_server_objects{std::make_shared()}; + std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>> m_rpc_system; // ThreadMap interface client, used to create a remote server thread when an @@ -679,8 +772,14 @@ ProxyClientBase::~ProxyClientBase() noexcept template ProxyServerBase::ProxyServerBase(std::shared_ptr impl, Connection& connection) - : m_impl(std::move(impl)), m_context(&connection) + : m_impl(std::move(impl)), m_context(&connection), m_server_objects(connection.m_server_objects) { + // Register this object with the connection's live-object tracker. This + // runs on the event loop thread, so it is ordered before any connection + // teardown (which also runs on the event loop thread): code that + // disconnects the connection and then calls Connection::waitDrained() is + // guaranteed to see this object. + m_server_objects->addServerObject(); MP_LOG(*m_context.loop, Log::Debug) << "Creating " << CxxTypeName(*this) << " " << this; assert(m_impl); } @@ -728,6 +827,15 @@ ProxyServerBase::~ProxyServerBase() } assert(m_context.cleanup_fns.empty()); MP_LOG(*m_context.loop, Log::Debug) << "Destroying " << CxxTypeName(*this) << " " << this; + // Deregister this object from the connection's live-object tracker, + // through the shared m_server_objects handle since m_context.connection + // may be dangling here (see comment above). Done at the end of the + // destructor so a zero count means destruction fully completed. Note that + // any m_impl destruction scheduled through addAsyncCleanup above is NOT + // covered by the tracker: it runs later on the async cleanup thread, so + // Connection::waitDrained() waits for server call bodies, not for + // m_impl destructors. + m_server_objects->removeServerObject(); } //! If the capnp interface defined a special "destroy" method, as described the diff --git a/include/mp/proxy.h b/include/mp/proxy.h index 2144b571..b02abfd8 100644 --- a/include/mp/proxy.h +++ b/include/mp/proxy.h @@ -20,6 +20,7 @@ namespace mp { class Connection; class EventLoop; +struct ServerObjectTracker; //! Mapping from capnp interface type to proxy client implementation (specializations are generated by //! proxy-codegen.cpp). template struct ProxyClient; // IWYU pragma: export @@ -172,6 +173,12 @@ struct ProxyServerBase : public virtual Interface_::Server * wrapped. */ std::shared_ptr m_impl; ProxyContext m_context; + //! Live-object tracker shared with this object's Connection, incremented + //! in the constructor and decremented in the destructor so shutdown code + //! can wait for a disconnected connection's server objects to drain. Held + //! via shared_ptr so it remains valid if this object (kept alive by an + //! in-flight call) outlives the Connection. See ServerObjectTracker. + std::shared_ptr m_server_objects; }; //! Customizable (through template specialization) base class which ProxyServer From 40cfb9332a1a47e4bc55f03aeb87787285f89975 Mon Sep 17 00:00:00 2001 From: Ryan Ofsky Date: Fri, 31 Jul 2026 15:40:26 -0400 Subject: [PATCH 8/8] test: cover draining in-flight server call after disconnect Add a deterministic mptest regression test for bitcoin/bitcoin#35845: hold a server method body in flight on a worker thread, call Connection::disconnect(), and assert that Connection::waitDrained() blocks until the body finishes and its server object is destroyed. Also covers destroying an already-disconnected connection (~Connection noticing disconnect() has run). Co-Authored-By: Claude Fable 5 Co-Authored-By: Enoch Azariah --- include/mp/proxy-io.h | 2 ++ test/mp/test/test.cpp | 75 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 54253c18..172a94c8 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -525,12 +525,14 @@ struct ServerObjectTracker void waitDrained() { Lock lock(m_mutex); + if (m_count != 0 && testing_hook_wait) testing_hook_wait(); m_cv.wait(lock.m_lock, [this]() MP_REQUIRES(m_mutex) { return m_count == 0; }); } mutable Mutex m_mutex; std::condition_variable m_cv; size_t m_count MP_GUARDED_BY(m_mutex){0}; + std::function testing_hook_wait; }; //! Object holding network & rpc state associated with either an incoming server diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 8bab6a06..de1f1604 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -597,6 +597,81 @@ KJ_TEST("Calling async IPC method, with server disconnect after cleanup") EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); } +KJ_TEST("Waiting for in-flight server call to finish after disconnect") +{ + // Regression test for bitcoin/bitcoin#35845. Disconnecting a connection + // cancels the KJ promise of an in-flight call, but a C++ server method + // body already dispatched to a worker thread runs to completion. Verify + // that Connection::waitDrained() blocks until such a body finishes and its + // server object is destroyed, so shutdown code can wait for a disconnected + // connection to become quiescent before freeing state the body accesses. + + std::promise body_started, release_body; + TestSetup setup; + ProxyClient* foo = setup.client.get(); + foo->initThreadMap(); + + // A server call body that signals when it starts and then blocks until the + // test releases it, so the in-flight state can be observed + // deterministically. + setup.server->m_impl->m_fn = [&] { + body_started.set_value(); + release_body.get_future().get(); + }; + + // Grab the server Connection object on the event loop thread before + // disconnecting. It stays valid until server_disconnect() destroys it + // below. + Connection* connection{nullptr}; + foo->m_context.loop->sync([&] { connection = setup.server->m_context.connection; }); + + // Invoke the async method on a separate thread so its body blocks there + // while this thread makes assertions. callFnAsync() takes an mp.Context, + // so its body runs on a worker thread via ProxyServer::post(). + std::thread call_thread([&] { + EXPECT_EXCEPTION(foo->callFnAsync(), "IPC client method call interrupted by disconnect."); + }); + body_started.get_future().get(); + + // The FooInterface server object is the connection's only counted server + // object, and its call body is executing. + KJ_EXPECT(connection->tracker()->pendingServerObjects() == 1); + + // Disconnect. This cancels the call's promise (the client above sees the + // disconnect error), but the body is still blocked on the worker thread, + // so its server object must still be alive. + foo->m_context.loop->sync([&] { connection->disconnect(); }); + KJ_EXPECT(connection->tracker()->pendingServerObjects() == 1); + + // A drain must block while the body runs and return only once it + // finishes, which is what Ipc::disconnectIncoming relies on during + // shutdown. + std::promise drain_waiting; + connection->tracker()->testing_hook_wait = [&] { drain_waiting.set_value(); }; + std::atomic drained{false}; + std::thread drain_thread([&] { + connection->tracker()->waitDrained(); + drained = true; + }); + + // Wait until waitDrained() has observed the live server object and is + // about to block, then verify it does not return while the body is blocked. + drain_waiting.get_future().get(); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + KJ_EXPECT(!drained); + + // Let the body finish; the drain should now complete. + release_body.set_value(); + drain_thread.join(); + KJ_EXPECT(drained); + KJ_EXPECT(connection->tracker()->pendingServerObjects() == 0); + call_thread.join(); + + // Destroy the drained connection. (~Connection notices disconnect() has + // already run and does not tear things down twice.) + setup.server_disconnect(); +} + KJ_TEST("Destroying ProxyClient<> with destroy method after peer disconnect") { // Regression test for bitcoin-core/libmultiprocess#219 where