shutdown bugfix: destroy RPC system before running cleanup callbacks

include/mp/proxy-io.h

@@ -333,7 +333,7 @@ class Connection
     LoggingErrorHandler m_error_handler{m_loop};
     kj::TaskSet m_on_disconnect{m_error_handler};
     ::capnp::TwoPartyVatNetwork m_network;
-    ::capnp::RpcSystem<::capnp::rpc::twoparty::VatId> m_rpc_system;
+    std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>> m_rpc_system;
 
     // ThreadMap interface client, used to create a remote server thread when an
     // client IPC call is being made for the first time from a new thread.
@@ -567,7 +567,7 @@ std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, int f
         auto stream =
             loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP);
         connection = std::make_unique<Connection>(loop, kj::mv(stream));
-        init_client = connection->m_rpc_system.bootstrap(ServerVatId().vat_id).castAs<InitInterface>();
+        init_client = connection->m_rpc_system->bootstrap(ServerVatId().vat_id).castAs<InitInterface>();
         Connection* connection_ptr = connection.get();
         connection->onDisconnect([&loop, connection_ptr] {
             loop.log() << "IPC client: unexpected network disconnect.";

src/mp/proxy.cpp

@@ -48,6 +48,12 @@ void LoggingErrorHandler::taskFailed(kj::Exception&& exception)
 
 Connection::~Connection()
 {
+    // Shut down RPC system first, since this will garbage collect Server
+    // objects that were not freed before the connection was closed, some of
+    // which may call addAsyncCleanup and add more cleanup callbacks which can
+    // run below.
+    m_rpc_system.reset();
+
     // ProxyClient cleanup handlers are in sync list, and ProxyServer cleanup
     // handlers are in the async list.
     //

test/mp/test/test.cpp

@@ -25,7 +25,7 @@ KJ_TEST("Call FooInterface methods")
 
         auto connection_client = std::make_unique<Connection>(loop, kj::mv(pipe.ends[0]));
         auto foo_client = std::make_unique<ProxyClient<messages::FooInterface>>(
-            connection_client->m_rpc_system.bootstrap(ServerVatId().vat_id).castAs<messages::FooInterface>(),
+            connection_client->m_rpc_system->bootstrap(ServerVatId().vat_id).castAs<messages::FooInterface>(),
             connection_client.get(), /* destroy_connection= */ false);
         foo_promise.set_value(std::move(foo_client));
         disconnect_client = [&] { loop.sync([&] { connection_client.reset(); }); };