Prevent crash on unclean disconnect if abandoned IPC call returns interface pointer

ryanofsky · ryanofsky · commit c436ae425066 · 2026-01-22T06:08:30.000-05:00
This is a potential fix for bitcoin/bitcoin#34250 which reports that bitcoin node crashes if a rust stratrumv2 mining client calls BlockTemplate.waitNext() and disconnects without waiting for a response from the call, and if mempool fees increased so the call returns a non-null interface BlockTemplate pointer. The node would crash in this case while trying to call MakeProxyServer on the returned BlockTemplate pointer, which would fail because MakeProxyServer would try to use a reference to the Connection object that had been deleted as a as a result of the disconnect. The fix works by: - Adding a Connection::m_canceler member variable and using it to cancel any IPC response promises that are pending when the connection is destroyed. - Updating type-context.h PassField() function to use promise.attach() as described https://capnproto.org/cxxrpc.html#cancellation to detect cancellation and set a ServerContext::m_cancelled variable. - Updating ServerCall to check the ServerContext::m_cancelled status after any C++ server method returns, and throw an exception if it is set. - Updating type-context.h PassField() function to deal with the exception by catching and logging it, and to deal with cancelled status by not trying to fulfill the cancelled promise.
diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h
@@ -48,6 +48,17 @@ struct ServerInvokeContext : InvokeContext
     ProxyServer& proxy_server;
     CallContext& call_context;
     int req;
+    //! For IPC methods that execute asynchronously, not on the event-loop
+    //! thread: lock preventing the event-loop thread from freeing the params or
+    //! results structs while the worker thread is reading params
+    //! (`call_context.getParams()`) or writing results
+    //! (`call_context.getResults()`).
+    Lock* cancel_lock{nullptr};
+    //! For IPC methods that execute asynchronously, not on the event-loop
+    //! thread, this is set to true if the IPC call was canceled by the client
+    //! or canceled by a disconnection. If the call runs on the event-loop
+    //! thread, it can't be canceled.
+    std::atomic<bool> canceled{false};
 
     ServerInvokeContext(ProxyServer& proxy_server, CallContext& call_context, int req)
         : InvokeContext{*proxy_server.m_context.connection}, proxy_server{proxy_server}, call_context{call_context}, req{req}
@@ -421,18 +432,21 @@ class Connection
     template <typename F>
     void onDisconnect(F&& f)
     {
-        // Add disconnect handler to local TaskSet to ensure it is cancelled and
+        // 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 cases where f deletes this Connection object.
+        // error in the typical case where f deletes this Connection object.
         m_on_disconnect.add(m_network.onDisconnect().then(
             [f = std::forward<F>(f), this]() mutable { m_loop->m_task_set->add(kj::evalLater(kj::mv(f))); }));
     }
 
     EventLoopRef m_loop;
     kj::Own<kj::AsyncIoStream> 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};
     ::capnp::TwoPartyVatNetwork m_network;
     std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>> m_rpc_system;
@@ -445,6 +459,11 @@ class Connection
     //! ThreadMap.makeThread) used to service requests to clients.
     ::capnp::CapabilityServerSet<Thread> m_threads;
 
+    //! Canceler for canceling promises that we want to discard when the
+    //! connection is destroyed. This is used to interrupt method calls that are
+    //! still executing at time of disconnection.
+    kj::Canceler m_canceler;
+
     //! Cleanup functions to run if connection is broken unexpectedly.  List
     //! will be empty if all ProxyClient are destroyed cleanly before the
     //! connection is destroyed.
@@ -696,9 +715,12 @@ template<typename T, typename Fn>
 kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
 {
     auto ready = kj::newPromiseAndFulfiller<void>(); // Signaled when waiter is idle again.
-    auto ret = m_thread_ready.then([this, fn = std::forward<Fn>(fn), ready_fulfiller = kj::mv(ready.fulfiller)]() mutable {
+    auto cancel_monitor_ptr = kj::heap<CancelMonitor>();
+    CancelMonitor& cancel_monitor = *cancel_monitor_ptr;
+    auto self = thisCap();
+    auto ret = m_thread_ready.then([this, self = std::move(self), fn = std::forward<Fn>(fn), ready_fulfiller = kj::mv(ready.fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr)]() mutable {
         auto result = kj::newPromiseAndFulfiller<T>(); // Signaled when fn() is called, with its return value.
-        bool posted = m_thread_context.waiter->post([this, fn = std::forward<Fn>(fn), ready_fulfiller = kj::mv(ready_fulfiller), result_fulfiller = kj::mv(result.fulfiller)]() mutable {
+        bool posted = m_thread_context.waiter->post([this, self = std::move(self), fn = std::forward<Fn>(fn), ready_fulfiller = kj::mv(ready_fulfiller), result_fulfiller = kj::mv(result.fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr)]() mutable {
             // Fullfill ready.promise now, as soon as the Waiter starts
             // executing this lambda, so the next ProxyServer<Thread>::post()
             // call can immediately call waiter->post() without needing to wait
@@ -713,8 +735,9 @@ kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
                 ready_fulfiller = nullptr;
             });
             std::optional<T> result_value;
-            kj::Maybe<kj::Exception> exception{kj::runCatchingExceptions([&]{ result_value.emplace(fn()); })};
-            m_loop->sync([&result_value, &exception, result_fulfiller = kj::mv(result_fulfiller)]() mutable {
+            kj::Maybe<kj::Exception> exception{kj::runCatchingExceptions([&]{ result_value.emplace(fn(*cancel_monitor_ptr)); })};
+            m_loop->sync([this, &result_value, &exception, self = kj::mv(self), result_fulfiller = kj::mv(result_fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr)]() mutable {
+                cancel_monitor_ptr = nullptr;
                 KJ_IF_MAYBE(e, exception) {
                     assert(!result_value);
                     result_fulfiller->reject(kj::mv(*e));
@@ -724,6 +747,7 @@ kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
                     result_value.reset();
                 }
                 result_fulfiller = nullptr;
+                m_loop->m_task_set->add(kj::evalLater([self = kj::mv(self)] {}));
             });
         });
         // Assert that calling Waiter::post did not fail. It could only return
@@ -732,7 +756,7 @@ kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
         // signaled, so this should never happen.
         assert(posted);
         return kj::mv(result.promise);
-    });
+    }).attach(kj::heap<CancelProbe>(cancel_monitor));
     m_thread_ready = kj::mv(ready.promise);
     return ret;
 }
diff --git a/include/mp/proxy-types.h b/include/mp/proxy-types.h
@@ -445,6 +445,15 @@ struct ServerCall
     template <typename ServerContext, typename... Args>
     decltype(auto) invoke(ServerContext& server_context, TypeList<>, Args&&... args) const
     {
+        // If cancel_lock is set, release it while executing the method, and
+        // reacquire it afterwards. The lock is needed to prevent params and
+        // response structs from being deleted by the event loop thread if the
+        // request is cancelled, so it is only needed before and after method
+        // execution. It is important to release the lock during execution the
+        // method can take an arbitrarily long time to return and the event loop
+        // will need the lock itself if the call is canceled.
+        if (server_context.cancel_lock) server_context.cancel_lock->m_lock.unlock();
+        KJ_DEFER(if (server_context.cancel_lock) server_context.cancel_lock->m_lock.lock());
         return ProxyServerMethodTraits<typename decltype(server_context.call_context.getParams())::Reads>::invoke(
             server_context,
             std::forward<Args>(args)...);
@@ -469,6 +478,12 @@ struct ServerRet : Parent
     void invoke(ServerContext& server_context, TypeList<>, Args&&... args) const
     {
         auto&& result = Parent::invoke(server_context, TypeList<>(), std::forward<Args>(args)...);
+        // If IPC request was canceled, there is no point continuing to execute.
+        // It's also important to stop executing because the connection may have
+        // been destroyed as described in
+        // https://github.com/bitcoin/bitcoin/issues/34250 and there would be a
+        // crash if execution continued.
+        if (server_context.canceled) throw InterruptException{"canceled"};
         auto&& results = server_context.call_context.getResults();
         InvokeContext& invoke_context = server_context;
         BuildField(TypeList<decltype(result)>(), invoke_context, Make<StructField, Accessor>(results),
diff --git a/include/mp/proxy.h b/include/mp/proxy.h
@@ -153,6 +153,7 @@ struct ProxyServerBase : public virtual Interface_::Server
     ProxyServerBase(std::shared_ptr<Impl> impl, Connection& connection);
     virtual ~ProxyServerBase();
     void invokeDestroy();
+    using Interface_::Server::thisCap;
 
     /**
      * Implementation pointer that may or may not be owned and deleted when this
diff --git a/include/mp/type-context.h b/include/mp/type-context.h
@@ -63,7 +63,14 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
     Context::Reader context_arg = Accessor::get(params);
     auto& server = server_context.proxy_server;
     int req = server_context.req;
-    auto invoke = [call_context = kj::mv(server_context.call_context), &server, req, fn, args...]() mutable {
+    // Keep a reference to the ProxyServer instance by assigning it to the self
+    // variable. ProxyServer instances are reference counted and if the client
+    // drops its reference and the IPC call is cancelled, this reference keeps
+    // the instance alive until the method finishes executing. The self variable
+    // needs to be destroyed on the event loop thread so it is freed in a sync()
+    // call below.
+    auto self = server.thisCap();
+    auto invoke = [self = kj::mv(self), call_context = kj::mv(server_context.call_context), &server, req, fn, args...](CancelMonitor& cancel_monitor) mutable {
                 MP_LOG(*server.m_context.loop, Log::Debug) << "IPC server executing request #" << req;
                 const auto& params = call_context.getParams();
                 Context::Reader context_arg = Accessor::get(params);
@@ -90,25 +97,58 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                     auto& request_threads = thread_context.request_threads;
                     ConnThread request_thread;
                     bool inserted;
+                    Mutex cancel_mutex;
+                    Lock cancel_lock{cancel_mutex};
+                    server_context.cancel_lock = &cancel_lock;
                     server.m_context.loop->sync([&] {
+                        // Detect request being canceled before it executes.
+                        if (cancel_monitor.m_canceled) {
+                            server_context.canceled = true;
+                            return;
+                        }
+                        // Detect request being canceled while it executes.
+                        assert(!cancel_monitor.m_on_cancel);
+                        cancel_monitor.m_on_cancel = [&server, &server_context, &cancel_mutex, req]() {
+                            MP_LOG(*server.m_context.loop, Log::Info) << "IPC server request #" << req << " canceled while executing.";
+                            server_context.canceled = true;
+                            // Wait for cancel_mutex to be released by the
+                            // execution thread before returning to the event
+                            // loop and letting parameters be deleted.
+                            Lock{cancel_mutex};
+                        };
+                        // Update requests_threads map if not canceled.
                         std::tie(request_thread, inserted) = SetThread(
                             GuardedRef{thread_context.waiter->m_mutex, request_threads}, server.m_context.connection,
                             [&] { return context_arg.getCallbackThread(); });
                     });
+                    if (server_context.canceled) {
+                        MP_LOG(*server.m_context.loop, Log::Info) << "IPC server request #" << req << " canceled before it could be executed";
+                        return call_context;
+                    }
 
                     // If an entry was inserted into the request_threads map,
                     // remove it after calling fn.invoke. If an entry was not
                     // inserted, one already existed, meaning this must be a
                     // recursive call (IPC call calling back to the caller which
                     // makes another IPC call), so avoid modifying the map.
                     const bool erase_thread{inserted};
-                    KJ_DEFER(if (erase_thread) {
+                    KJ_DEFER(
+                        // Release the cancel lock before calling loop->sync and
+                        // waiting for the event loop thread because if a
+                        // cancellation happened it will run the  on_cancel
+                        // callback, above. It's safe to release cancel_lock
+                        // lock at this point because the fn.invoke() call below
+                        // will be finished and no longer accessing the
+                        // parameters or response structs.
+                        cancel_lock.m_lock.unlock();
                         // Erase the request_threads entry on the event loop
                         // thread with loop->sync(), so if the connection is
                         // broken there is not a race between this thread and
                         // the disconnect handler trying to destroy the thread
                         // client object.
                         server.m_context.loop->sync([&] {
+                            auto self_dispose{kj::mv(self)};
+                            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
@@ -120,9 +160,19 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                                 Lock lock(thread_context.waiter->m_mutex);
                                 removed = request_threads.extract(server.m_context.connection);
                             }
+                            }
                         });
-                    });
-                    fn.invoke(server_context, args...);
+                    );
+                    KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]{
+                        try {
+                            fn.invoke(server_context, args...);
+                        } catch (const InterruptException& e) {
+                            MP_LOG(*server.m_context.loop, Log::Info) << "IPC server request #" << req << " interrupted (" << e.what() << ")";
+                        }
+                    })) {
+                        MP_LOG(*server.m_context.loop, Log::Error) << "IPC server request #" << req << " uncaught exception.";
+                        throw exception;
+                    }
                 }
                 return call_context;
             };
@@ -131,7 +181,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
     // be a local Thread::Server object, but it needs to be looked up
     // asynchronously with getLocalServer().
     auto thread_client = context_arg.getThread();
-    return server.m_context.connection->m_threads.getLocalServer(thread_client)
+    auto result = server.m_context.connection->m_threads.getLocalServer(thread_client)
         .then([&server, invoke = kj::mv(invoke), req](const kj::Maybe<Thread::Server&>& perhaps) mutable {
             // Assuming the thread object is found, pass it a pointer to the
             // `invoke` lambda above which will invoke the function on that
@@ -147,6 +197,12 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                 throw std::runtime_error("invalid thread handle");
             }
         });
+    // Use connection m_canceler object to cancel the result promise if the
+    // connection is destroyed. (By default. Cap'n Proto does not cancel
+    // requests on disconnect, since it's possible clients could send requests
+    // and disconnect without waiting for the results and not want those
+    // requests to be canceled.)
+    return server.m_context.connection->m_canceler.wrap(kj::mv(result));
 }
 } // namespace mp
 
diff --git a/include/mp/util.h b/include/mp/util.h
@@ -9,6 +9,7 @@
 #include <cassert>
 #include <cstddef>
 #include <cstring>
+#include <exception>
 #include <functional>
 #include <kj/string-tree.h>
 #include <mutex>
@@ -241,6 +242,64 @@ inline char* CharCast(unsigned char* c) { return (char*)c; }
 inline const char* CharCast(const char* c) { return c; }
 inline const char* CharCast(const unsigned char* c) { return (const char*)c; }
 
+//! Exception that can be thrown from code executing an IPC call that is interrupted.
+struct InterruptException final : std::exception {
+    explicit InterruptException(std::string message) : m_message(std::move(message)) {}
+    const char* what() const noexcept override { return m_message.c_str(); }
+    std::string m_message;
+};
+
+class CancelProbe;
+
+//! Helper class that detects when a promise is canceled. Used to detect
+//! canceled requests and prevent potential crashes on unclean disconnects.
+//!
+//! In the future, this could also be used to support a way for wrapped C++
+//! methods to detect cancellation (like approach #4 in
+//! https://github.com/bitcoin/bitcoin/issues/33575).
+class CancelMonitor
+{
+public:
+    inline ~CancelMonitor();
+    inline void setCancelled(CancelProbe& probe);
+
+    bool m_canceled{false};
+    std::function<void()> m_on_cancel;
+    CancelProbe* m_probe{nullptr};
+};
+
+//! Helper object to attach to a promise and update a CancelMonitor.
+class CancelProbe
+{
+public:
+    CancelProbe(CancelMonitor& monitor) : m_monitor(&monitor)
+    {
+        assert(!monitor.m_probe);
+        monitor.m_probe = this;
+    }
+    ~CancelProbe()
+    {
+        if (m_monitor) m_monitor->setCancelled(*this);
+    }
+    CancelMonitor* m_monitor;
+};
+
+CancelMonitor::~CancelMonitor()
+{
+    if (m_probe) {
+        assert(m_probe->m_monitor == this);
+        m_probe->m_monitor = nullptr;
+        m_probe = nullptr;
+    }
+}
+
+void CancelMonitor::setCancelled(CancelProbe& probe)
+{
+    assert(m_probe == &probe);
+    m_canceled = true;
+    if (m_on_cancel) m_on_cancel();
+    m_probe = nullptr;
+}
 } // namespace mp
 
 #endif // MP_UTIL_H
diff --git a/src/mp/gen.cpp b/src/mp/gen.cpp
@@ -211,6 +211,7 @@ static void Generate(kj::StringPtr src_prefix,
     cpp_server << "#include <kj/async.h>\n";
     cpp_server << "#include <kj/common.h>\n";
     cpp_server << "#include <kj/exception.h>\n";
+    cpp_server << "#include <kj/tuple.h>\n";
     cpp_server << "#include <mp/proxy.h>\n";
     cpp_server << "#include <mp/util.h>\n";
     cpp_server << "#include <" << PROXY_TYPES << ">\n";
diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp
@@ -87,6 +87,10 @@ Connection::~Connection()
     // event loop thread, and if there was a remote disconnect, this is called
     // by an 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.
+    m_canceler.cancel("Interrupted by disconnect");
+
     // Shut down RPC system first, since this will garbage collect any
     // ProxyServer objects that were not freed before the connection was closed.
     // Typically all ProxyServer objects associated with this connection will be
