type-context: fix async disconnect race condition found by antithesis

include/mp/proxy-io.h

@@ -12,6 +12,7 @@
 
 #include <capnp/rpc-twoparty.h>
 
+#include <any>
 #include <assert.h>
 #include <algorithm>
 #include <condition_variable>
@@ -374,6 +375,12 @@ class EventLoop
 
     //! Hook called on the event loop thread when a client has disconnected.
     std::function<void()> testing_hook_disconnected;
+
+    //! 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
+    //! requiring a dedicated hook for each one.
+    std::function<void(std::any)> testing_hook_misc;
 };
 
 //! Single element task queue used to handle recursive capnp calls. (If the

include/mp/type-context.h

@@ -131,6 +131,18 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
             std::tie(request_thread, inserted) = SetThread(
                 GuardedRef{thread_context.waiter->m_mutex, request_threads}, server.m_context.connection,
                 [&] { return Accessor::get(call_context.getParams()).getCallbackThread(); });
+            // Initialize the request's results struct here on the event loop
+            // thread, so later getResults() calls on the execution thread
+            // return the response capnp caches on first use instead of
+            // reading Cap'n Proto connection state, which is unsafe off the
+            // event loop thread because RpcConnectionState::disconnect()
+            // overwrites it there without synchronization on an abrupt remote
+            // disconnect (bitcoin-core/libmultiprocess#348). After this call,
+            // the only call_context state accessed by the execution thread is
+            // the params reader and the results struct allocated here; any
+            // future change accessing other call_context state needs to move
+            // that access to the event loop thread the same way.
+            call_context.getResults();
         });
 
         // If an entry was inserted into the request_threads map,

test/mp/test/foo-types.h

@@ -9,6 +9,7 @@
 #include <mp/proxy-types.h>
 
 // IWYU pragma: begin_exports
+#include <any>
 #include <capnp/common.h>
 #include <cstddef>
 #include <mp/test/foo.capnp.h>
@@ -75,6 +76,8 @@ inline void CustomBuildMessage(InvokeContext& invoke_context,
                         const test::FooMessage& src,
                         test::messages::FooMessage::Builder&& builder)
 {
+    const auto& hook{invoke_context.connection.m_loop->testing_hook_misc};
+    if (hook) hook("build FooMessage");
     builder.setMessage(src.message + " build");
 }
 

test/mp/test/foo.capnp

@@ -38,6 +38,7 @@ interface FooInterface $Proxy.wrap("mp::test::FooImplementation") {
     callIntFnAsync @21 (context :Proxy.Context, arg :Int32) -> (result :Int32);
     passDataPointers @22 (arg :List(Data)) -> (result :List(Data));
     listBars @25 (context :Proxy.Context, n :Int32) -> (result :List(BarInterface));
+    callMessageAsync @26 (context :Proxy.Context) -> (result :FooMessage);
 }
 
 interface FooInit $Proxy.wrap("mp::test::FooInit") {

test/mp/test/foo.h

@@ -124,6 +124,7 @@ class FooImplementation
     void callFn() { assert(m_fn); m_fn(); }
     void callFnAsync() { assert(m_fn); m_fn(); }
     int callIntFnAsync(int arg) { assert(m_int_fn); return m_int_fn(arg); }
+    FooMessage callMessageAsync() { assert(m_fn); m_fn(); return {}; }
     std::function<void()> m_fn;
     std::function<int(int)> m_int_fn;
 };

test/mp/test/test.cpp

@@ -5,6 +5,7 @@
 #include <mp/test/foo.capnp.h>
 #include <mp/test/foo.capnp.proxy.h>
 
+#include <any>
 #include <atomic>
 #include <capnp/capability.h>
 #include <capnp/rpc.h>
@@ -79,6 +80,7 @@ class TestSetup
 public:
     std::function<void()> server_disconnect;
     std::function<void()> server_disconnect_later;
+    std::function<void()> server_on_disconnect;
     std::function<void()> client_disconnect;
     std::promise<std::unique_ptr<ProxyClient<messages::FooInterface>>> client_promise;
     std::unique_ptr<ProxyClient<messages::FooInterface>> client;
@@ -109,8 +111,14 @@ class TestSetup
                   loop.m_task_set->add(kj::evalLater([&] { server_connection.reset(); }));
               };
               // Set handler to destroy the server when the client disconnects. This
-              // is ignored if server_disconnect() is called instead.
-              server_connection->onDisconnect([&] { server_connection.reset(); });
+              // is ignored if server_disconnect() is called instead. Tests can
+              // assign server_on_disconnect to override the default behavior of
+              // destroying the server connection as soon as the disconnect is
+              // detected (in which case they need to destroy it themselves,
+              // e.g. by calling server_disconnect(), so the event loop can
+              // exit).
+              server_on_disconnect = [&] { server_connection.reset(); };
+              server_connection->onDisconnect([&] { server_on_disconnect(); });
 
               auto client_connection = std::make_unique<Connection>(loop, kj::mv(pipe.ends[1]));
               auto client_proxy = std::make_unique<ProxyClient<messages::FooInterface>>(
@@ -368,6 +376,98 @@ KJ_TEST("Calling IPC method, disconnecting and blocking during the call")
     signal.set_value();
 }
 
+KJ_TEST("Calling async IPC method with a remote disconnect while results are built")
+{
+    // Regression test for bitcoin-core/libmultiprocess#348, a data race
+    // reported by ThreadSanitizer where a server worker thread called
+    // call_context.getResults() while the event loop thread was tearing down
+    // Cap'n Proto connection state (capnp::_::RpcConnectionState::disconnect()
+    // overwriting the RpcConnectionState::connection field) after an abrupt
+    // remote disconnect.
+    //
+    // The test makes an async IPC call (callMessageAsync) whose method body
+    // just signals the main thread. When the worker thread returns from the
+    // method body, it calls call_context.getResults(), reading the connection
+    // state, and starts serializing the FooMessage result, where the
+    // testing_hook_misc hook set below makes it sleep. Meanwhile the main
+    // thread destroys the client connection, and the event loop thread
+    // processes the resulting EOF, running RpcConnectionState::disconnect()
+    // and overwriting the connection state the worker thread just read, with
+    // no synchronization between the two accesses.
+    //
+    // Two details are essential for ThreadSanitizer to detect the race:
+    //
+    // - There must be no synchronization between the worker thread's
+    //   getResults() call and the event loop thread's disconnect processing.
+    //   The worker signals the main thread *before* getResults() and sleeps
+    //   (sleeping creates no happens-before edge) across the disconnect, so
+    //   the racing accesses are not ordered by any of the test's own
+    //   synchronization.
+    //
+    // - The worker thread's read should come *before* the event loop thread's
+    //   write, close in time. In the write-then-read order (e.g. method
+    //   returning long after the disconnect), the race exists too, but
+    //   ThreadSanitizer usually misses it: right after disconnect() writes the
+    //   connection field, the loop thread's teardown destructors re-read it
+    //   many times (~ImportClient etc. check connection.is<Connected>() to
+    //   decide whether to send messages), evicting the write from the
+    //   per-granule shadow history before a late reader comes along.
+    //
+    // The server Connection object is deliberately kept alive during all this
+    // by overriding server_on_disconnect: destroying it would cancel the
+    // in-flight request (Connection::~Connection calls m_canceler.cancel(),
+    // 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.
+
+    TestSetup setup{/*client_owns_connection=*/false};
+    ProxyClient<messages::FooInterface>* foo = setup.client.get();
+    KJ_EXPECT(foo->add(1, 2) == 3);
+    foo->initThreadMap();
+
+    // Keep the server Connection object alive when the disconnect is detected
+    // so the in-flight request is not canceled (see comment above). The
+    // connection is destroyed at the end of the test instead.
+    setup.server_on_disconnect = [] {};
+
+    // Signaled by the worker thread when the method body runs, just before it
+    // returns and the worker calls getResults() and serializes the results.
+    std::promise<void> fn_called;
+    setup.server->m_impl->m_fn = [&] { fn_called.set_value(); };
+
+    // Keep the worker thread inside the results-building step, without
+    // synchronizing, while the event loop thread processes the disconnect.
+    EventLoop& loop = *setup.server->m_context.connection->m_loop;
+    loop.testing_hook_misc = [](std::any arg) {
+        if (const char* const* tag{std::any_cast<const char*>(&arg)};
+            tag && std::string_view{*tag} == "build FooMessage") {
+            std::this_thread::sleep_for(std::chrono::milliseconds(200));
+        }
+    };
+
+    // Signaled by the worker thread when it is completely done with the
+    // request, including serializing the results.
+    std::promise<void> request_done;
+    loop.testing_hook_async_request_done = [&] { request_done.set_value(); };
+
+    // Make the IPC call from a separate thread so this thread can trigger the
+    // client disconnect while the call is executing.
+    std::thread caller{[&] {
+        EXPECT_EXCEPTION(foo->callMessageAsync(), "IPC client method call interrupted by disconnect.");
+    }};
+
+    fn_called.get_future().get();
+    setup.client_disconnect();
+    caller.join();
+
+    // Wait for the worker thread to finish the request, then tear down the
+    // server connection that was deliberately kept alive above, so the event
+    // loop is able to exit.
+    request_done.get_future().get();
+    setup.server_disconnect();
+}
+
 KJ_TEST("Worker thread destroyed before it is initialized")
 {
     // Regression test for bitcoin/bitcoin#34711, bitcoin/bitcoin#34756 where a