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

ryanofsky · ryanofsky · commit 92a2c5f66282 · 2026-01-20T13:23:59.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,11 @@ struct ServerInvokeContext : InvokeContext
     ProxyServer& proxy_server;
     CallContext& call_context;
     int req;
+    //! If the IPC method executes asynchronously (not on the event loop thread)
+    //! and the IPC call was cancelled by the client, or cancelled by a
+    //! disconnection, this is set to true. If the call runs on the event loop
+    //! thread, it can't be cancelled.
+    std::atomic<bool> cancelled{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}
@@ -422,14 +427,17 @@ class Connection
         // 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;
@@ -442,6 +450,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.
@@ -693,20 +706,26 @@ 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::move(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::move(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 {
             m_loop->sync([ready_fulfiller = kj::mv(ready_fulfiller)]() mutable { ready_fulfiller->fulfill(); });
             std::optional<T> result;
-            kj::Maybe<kj::Exception> exception{kj::runCatchingExceptions([&]{ result.emplace(fn()); })};
-            m_loop->sync([&result, &exception, result_fulfiller = kj::mv(result_fulfiller)]() mutable {
+            kj::Maybe<kj::Exception> exception{kj::runCatchingExceptions([&]{ result.emplace(fn(*cancel_monitor_ptr)); })};
+            m_loop->sync([this, &result, &exception, self = kj::mv(self), result_fulfiller = kj::mv(result_fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr)]() mutable {
+                auto result_fulfiller_dispose{kj::mv(result_fulfiller)};
+                auto cancel_monitor_dispose{kj::mv(cancel_monitor_ptr)};
                 KJ_IF_MAYBE(e, exception) {
                     assert(!result);
-                    result_fulfiller->reject(kj::mv(*e));
+                    result_fulfiller_dispose->reject(kj::mv(*e));
                 } else {
                     assert(result);
-                    result_fulfiller->fulfill(kj::mv(*result));
+                    result_fulfiller_dispose->fulfill(kj::mv(*result));
                 }
+                m_loop->m_task_set->add(kj::evalLater([self = kj::mv(self)] {}));
             });
         });
         // Assert that calling Waiter::post did not fail. It could only return
@@ -715,7 +734,7 @@ kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
         // is 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
@@ -469,6 +469,16 @@ 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 cancelled, 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.
+        // TODO: Note this detection is racy because cancellation could happen
+        // after this check. However, fixing this would require changing the
+        // definition of the InvokeContext struct and updating a lot of code, so
+        // this can be done in a followup.
+        if (server_context.cancelled) throw InterruptException{"cancelled"};
         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,8 @@ 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 {
+    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);
@@ -91,6 +92,15 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                     ConnThread request_thread;
                     bool inserted;
                     server.m_context.loop->sync([&] {
+                        // Detect request being cancelled before or while it executes.
+                        if (cancel_monitor.m_cancelled) MP_LOG(*server.m_context.loop, Log::Raise) << "IPC server request #" << req << " cancelled before it could be executed";
+                        assert(!cancel_monitor.m_on_cancel);
+                        cancel_monitor.m_on_cancel = [&server, &server_context, req]() {
+                            MP_LOG(*server.m_context.loop, Log::Error) << "IPC server request #" << req << " cancelled while executing.";
+                            server_context.cancelled = true;
+                        };
+
+                        // Update requests_threads map if not cancelled.
                         std::tie(request_thread, inserted) = SetThread(
                             GuardedRef{thread_context.waiter->m_mutex, request_threads}, server.m_context.connection,
                             [&] { return context_arg.getCallbackThread(); });
@@ -102,13 +112,15 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                     // 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(
                         // 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 +132,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::Error) << "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 +153,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 +169,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 cancelled.)
+    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>
@@ -238,6 +239,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 cancelled. Used to detect
+//! cancelled 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_cancelled{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_cancelled = 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
