diff --git a/src/ipc/libmultiprocess/test/mp/test/foo-types.h b/src/ipc/libmultiprocess/test/mp/test/foo-types.h index b96eabfcfb2..0abb779816b 100644 --- a/src/ipc/libmultiprocess/test/mp/test/foo-types.h +++ b/src/ipc/libmultiprocess/test/mp/test/foo-types.h @@ -28,6 +28,7 @@ #include #include #include +#include #include #include // IWYU pragma: end_exports @@ -72,6 +73,9 @@ inline void CustomBuildMessage(InvokeContext& invoke_context, const test::FooMessage& src, test::messages::FooMessage::Builder&& builder) { + // Reproduction for AR-14: a field builder that throws while the + // request is being built. + if (src.message == "throw") throw std::runtime_error("FooMessage build throw"); builder.setMessage(src.message + " build"); } diff --git a/src/ipc/libmultiprocess/test/mp/test/foo.capnp b/src/ipc/libmultiprocess/test/mp/test/foo.capnp index 9e6213fd768..473070be118 100644 --- a/src/ipc/libmultiprocess/test/mp/test/foo.capnp +++ b/src/ipc/libmultiprocess/test/mp/test/foo.capnp @@ -37,6 +37,7 @@ interface FooInterface $Proxy.wrap("mp::test::FooImplementation") { callFnAsync @18 (context :Proxy.Context) -> (); callIntFnAsync @21 (context :Proxy.Context, arg :Int32) -> (result :Int32); passDataPointers @22 (arg :List(Data)) -> (result :List(Data)); + makeCallback @25 () -> (result :FooCallback); } interface FooCallback $Proxy.wrap("mp::test::FooCallback") { diff --git a/src/ipc/libmultiprocess/test/mp/test/foo.h b/src/ipc/libmultiprocess/test/mp/test/foo.h index 779c8db1b03..7c63041d40c 100644 --- a/src/ipc/libmultiprocess/test/mp/test/foo.h +++ b/src/ipc/libmultiprocess/test/mp/test/foo.h @@ -60,6 +60,21 @@ public: virtual int call(int arg) = 0; }; +//! FooCallback implementation owned by the server (returned from +//! makeCallback), used by the reproduction tests for calls after destroy. +class OwnedCallback : public FooCallback +{ +public: + explicit OwnedCallback(std::function hook) : m_hook(std::move(hook)) {} + int call(int arg) override + { + if (m_hook) m_hook(); + return m_value + arg; + } + std::function m_hook; + int m_value{100}; +}; + class ExtendedCallback : public FooCallback { public: @@ -76,7 +91,10 @@ public: FooStruct pass(FooStruct foo) { return foo; } void raise(FooStruct foo) { throw foo; } void initThreadMap() {} - int callback(FooCallback& callback, int arg) { return callback.call(arg); } + int callback(FooCallback& callback, int arg) { if (m_callback_ref_hook) m_callback_ref_hook(callback); return callback.call(arg); } + std::function m_callback_ref_hook; + std::unique_ptr makeCallback() { return std::make_unique(m_owned_callback_hook); } + std::function m_owned_callback_hook; int callbackUnique(std::unique_ptr callback, int arg) { return callback->call(arg); } int callbackShared(std::shared_ptr callback, int arg) { return callback->call(arg); } // NOLINT(performance-unnecessary-value-param) void saveCallback(std::shared_ptr callback) { m_callback = std::move(callback); } diff --git a/src/ipc/libmultiprocess/test/mp/test/spawn_tests.cpp b/src/ipc/libmultiprocess/test/mp/test/spawn_tests.cpp index a14e50e2a30..43d761b44c3 100644 --- a/src/ipc/libmultiprocess/test/mp/test/spawn_tests.cpp +++ b/src/ipc/libmultiprocess/test/mp/test/spawn_tests.cpp @@ -13,6 +13,9 @@ #include #include #include +#include +#include +#include #include #include #include @@ -110,3 +113,85 @@ KJ_TEST("SpawnProcess does not run callback in child") KJ_EXPECT(exited, "Timeout waiting for child process to exit"); KJ_EXPECT(WIFEXITED(status) && WEXITSTATUS(status) == 0); } + +namespace { +//! Spawn `sh -c 'test -e /proc/$$/fd/N'` and return whether descriptor N was +//! open in the spawned process. +bool FdOpenInSpawnedChild(int target_fd) +{ + int pid{-1}; + const int fd{mp::SpawnProcess(pid, [&](int) -> std::vector { + return {"sh", "-c", "test -e /proc/$$/fd/" + std::to_string(target_fd)}; + })}; + ::close(fd); + const int status{mp::WaitProcess(pid)}; + return WIFEXITED(status) && WEXITSTATUS(status) == 0; +} +} // namespace + +KJ_TEST("repro AR-15: SpawnProcess closes the highest descriptor below RLIMIT_NOFILE") +{ + struct rlimit lim; + KJ_REQUIRE(getrlimit(RLIMIT_NOFILE, &lim) == 0); + const int top{static_cast(lim.rlim_cur - 1)}; + const int below{static_cast(lim.rlim_cur - 2)}; + const int devnull{::open("/dev/null", O_RDONLY)}; // no O_CLOEXEC + KJ_REQUIRE(devnull >= 0); + KJ_REQUIRE(::dup2(devnull, top) == top); + KJ_REQUIRE(::dup2(devnull, below) == below); + const bool leaked_below{FdOpenInSpawnedChild(below)}; + const bool leaked_top{FdOpenInSpawnedChild(top)}; + ::close(top); + ::close(below); + ::close(devnull); + KJ_LOG(WARNING, "rlim_cur", lim.rlim_cur, "fd", below, leaked_below, "fd", top, leaked_top); + KJ_EXPECT(!leaked_below); + KJ_EXPECT(!leaked_top, "descriptor rlim_cur - 1 inherited by spawned process"); +} + +KJ_TEST("repro AR-15: SpawnProcess closes descriptors above a lowered RLIMIT_NOFILE") +{ + struct rlimit saved; + KJ_REQUIRE(getrlimit(RLIMIT_NOFILE, &saved) == 0); + const int devnull{::open("/dev/null", O_RDONLY)}; // no O_CLOEXEC + KJ_REQUIRE(::dup2(devnull, 100) == 100); + struct rlimit lowered{saved}; + lowered.rlim_cur = 64; + KJ_REQUIRE(setrlimit(RLIMIT_NOFILE, &lowered) == 0); + const bool leaked{FdOpenInSpawnedChild(100)}; + KJ_REQUIRE(setrlimit(RLIMIT_NOFILE, &saved) == 0); + ::close(100); + ::close(devnull); + KJ_LOG(WARNING, "fd 100 with soft limit 64 inherited", leaked); + KJ_EXPECT(!leaked, "descriptor above lowered limit inherited by spawned process"); +} + +KJ_TEST("repro mp#311: fork+exec concurrent with SpawnProcess does not inherit the IPC socket") +{ + // Stand in for another thread running a shell hook (like -blocknotify, + // which uses std::system) while SpawnProcess is between socketpair() and + // its close(fds[0]) after fork(): run the hook from the fd_to_args + // callback, which SpawnProcess calls in that window. The hook leaves a + // background process running for 3 seconds. + int pid{-1}; + const int fd{mp::SpawnProcess(pid, [&](int) -> std::vector { + KJ_REQUIRE(std::system("sleep 3 &") == 0); + return {"true"}; + })}; + const int flags{::fcntl(fd, F_GETFD)}; + const auto start{std::chrono::steady_clock::now()}; + KJ_REQUIRE(mp::WaitProcess(pid) == 0); + // The spawned process has exited, so the parent should read EOF at once. + struct pollfd pfd{fd, POLLIN, 0}; + char byte; + const int ready{::poll(&pfd, 1, 10000)}; + const bool eof{ready == 1 && ::read(fd, &byte, 1) == 0}; + const auto elapsed{std::chrono::duration_cast(std::chrono::steady_clock::now() - start)}; + // A process started after SpawnProcess returned also inherits the + // parent's end unless it is close-on-exec. + const bool later_inherits{std::system(("test -e /proc/$$/fd/" + std::to_string(fd)).c_str()) == 0}; + ::close(fd); + KJ_LOG(WARNING, "parent fd FD_CLOEXEC", (flags & FD_CLOEXEC) != 0, "EOF after ms", elapsed.count(), eof, "later std::system inherits parent end", later_inherits); + KJ_EXPECT(elapsed < std::chrono::milliseconds{1000}, "EOF delayed until the unrelated process exited"); + KJ_EXPECT(!later_inherits, "parent IPC socket inherited by a later std::system child"); +} diff --git a/src/ipc/libmultiprocess/test/mp/test/test.cpp b/src/ipc/libmultiprocess/test/mp/test/test.cpp index 7a0940c8de7..06fef5df4e3 100644 --- a/src/ipc/libmultiprocess/test/mp/test/test.cpp +++ b/src/ipc/libmultiprocess/test/mp/test/test.cpp @@ -12,6 +12,8 @@ #include #include #include +#include +#include #include #include #include @@ -86,7 +88,8 @@ public: //! not start until the other members are initialized. std::thread thread; - TestSetup(bool client_owns_connection = true) + TestSetup(bool client_owns_connection = true, + std::function>(Connection&)> make_server = {}) : thread{[&] { EventLoop loop("mptest", [](mp::LogMessage log) { // Info logs are not printed by default, but will be shown with `mptest --verbose` @@ -97,7 +100,7 @@ public: auto server_connection = std::make_unique(loop, kj::mv(pipe.ends[0]), [&](Connection& connection) { - auto server_proxy = kj::heap>( + auto server_proxy = make_server ? make_server(connection) : kj::heap>( std::make_shared(), connection); server = server_proxy; return capnp::Capability::Client(kj::mv(server_proxy)); @@ -577,5 +580,267 @@ KJ_TEST("Call async IPC method without thread or pool errors correctly") KJ_EXPECT(error_thrown); } + +//! Reproduction tests for open stability findings (ipcbugs survey, +//! 2026-09-26). Each test checks for the correct behavior, so it fails (or +//! crashes, or reports under ASan) while the bug is present. + +//! Get the request and callback Thread clients the current thread uses for +//! the connection, after at least one call with a Context has been made. +static void GetThreads(ProxyClient& foo, Thread::Client*& request_thread, Thread::Client*& callback_thread) +{ + ThreadContext& tc{g_thread_context}; + foo.m_context.loop->sync([&] { + Lock lock(tc.waiter->m_mutex); + callback_thread = &tc.callback_threads.at(foo.m_context.connection)->m_client; + request_thread = &tc.request_threads.at(foo.m_context.connection)->m_client; + }); +} + +KJ_TEST("repro AR-14: clientInvoke returns when building the request throws") +{ + TestSetup setup; + ProxyClient* foo = setup.client.get(); + auto result = std::make_shared>(); + auto future = result->get_future(); + std::thread caller([foo, result] { + try { + FooMessage message; + message.message = "throw"; + foo->passMessage(message); + result->set_value("returned normally"); + } catch (const std::exception& e) { + result->set_value(std::string{"threw: "} + e.what()); + } + }); + const bool finished{future.wait_for(std::chrono::seconds{5}) == std::future_status::ready}; + if (finished) { + KJ_LOG(WARNING, "passMessage outcome", future.get()); + caller.join(); + } else { + KJ_LOG(WARNING, "passMessage still blocked in clientInvoke after 5 seconds"); + caller.detach(); + } + KJ_EXPECT(finished, "clientInvoke hung after CustomBuildField threw"); +} + +KJ_TEST("repro AR-17: call after destroy on an owned server object") +{ + TestSetup setup; + ProxyClient* foo = setup.client.get(); + foo->initThreadMap(); + std::unique_ptr callback{foo->makeCallback()}; + KJ_EXPECT(callback->call(1) == 101); + Thread::Client *request_thread, *callback_thread; + GetThreads(*foo, request_thread, callback_thread); + + // Send destroy and then call over the same capability, as a client that + // does not use the generated C++ ProxyClient could. + auto& client{dynamic_cast&>(*callback).m_client}; + std::promise outcome; + foo->m_context.loop->sync([&] { + auto destroy{client.destroyRequest()}; + auto destroy_context{destroy.initContext()}; + destroy_context.setThread(*request_thread); + destroy_context.setCallbackThread(*callback_thread); + foo->m_context.loop->m_task_set->add(destroy.send().then([&](auto&&) { + auto call{client.callRequest()}; + auto call_context{call.initContext()}; + call_context.setThread(*request_thread); + call_context.setCallbackThread(*callback_thread); + call.setArg(2); + return call.send().then( + [&](auto&& results) { outcome.set_value("call after destroy returned " + std::to_string(results.getResult())); }, + [&](kj::Exception&& e) { outcome.set_value(std::string{"call after destroy failed: "} + e.getDescription().cStr()); }); + })); + }); + const std::string result{outcome.get_future().get()}; + KJ_LOG(WARNING, result); + KJ_EXPECT(result.starts_with("call after destroy failed")); +} + +KJ_TEST("repro AR-17: destroy while a call is running on another thread") +{ + TestSetup setup; + ProxyClient* foo = setup.client.get(); + foo->initThreadMap(); + std::atomic block{false}; + std::promise entered; + std::promise release; + std::shared_future release_future{release.get_future().share()}; + setup.server->m_impl->m_owned_callback_hook = [&block, &entered, release_future] { + if (block.exchange(false)) { + entered.set_value(); + release_future.wait(); + } + }; + std::unique_ptr callback{foo->makeCallback()}; + KJ_EXPECT(callback->call(1) == 101); + Thread::Client *request_thread, *callback_thread; + GetThreads(*foo, request_thread, callback_thread); + + // Thread A: a call that blocks inside the implementation. + block = true; + std::promise call_result; + std::thread caller([&] { call_result.set_value(callback->call(2)); }); + entered.get_future().wait(); + + // Thread B: destroy on a second server thread while the call is running. + auto& client{dynamic_cast&>(*callback).m_client}; + std::promise destroyed; + foo->m_context.loop->sync([&] { + auto make_thread{foo->m_context.connection->m_thread_map.makeThreadRequest()}; + make_thread.setName("destroyer"); + foo->m_context.loop->m_task_set->add(make_thread.send().then([&](auto&& results) { + Thread::Client destroy_thread{results.getResult()}; + auto destroy{client.destroyRequest()}; + auto destroy_context{destroy.initContext()}; + destroy_context.setThread(destroy_thread); + destroy_context.setCallbackThread(*callback_thread); + return destroy.send().then([&, destroy_thread](auto&&) { destroyed.set_value(); }); + })); + }); + destroyed.get_future().wait(); + KJ_LOG(WARNING, "destroy returned while call was still running"); + release.set_value(); + const int result{call_result.get_future().get()}; + caller.join(); + KJ_LOG(WARNING, "call running during destroy returned", result); + KJ_EXPECT(result == 102); +} + +KJ_TEST("repro AR-02: generated server retaining an Impl& capability") +{ + class Callback : public FooCallback + { + public: + int call(int arg) override { return arg + m_offset; } + int m_offset{1000}; + }; + + TestSetup setup; + ProxyClient* foo = setup.client.get(); + foo->initThreadMap(); + EventLoop& loop{*setup.server->m_context.loop}; + std::optional retained; + // The server side keeps a copy of the capability it received for the + // FooCallback& argument, as a peer not bound by the generated code could. + setup.server->m_impl->m_callback_ref_hook = [&](FooCallback& cb) { + loop.sync([&] { retained.emplace(dynamic_cast&>(cb).m_client); }); + }; + { + auto callback{std::make_unique()}; + KJ_EXPECT(foo->callback(*callback, 1) == 1001); + } // referent destroyed here; the client-side ProxyServer still points to it + KJ_REQUIRE(retained.has_value()); + loop.sync([&] { + setup.server->m_impl->m_callback = std::make_shared>( + *retained, setup.server->m_context.connection, /*destroy_connection=*/false); + retained.reset(); + }); + const int result{foo->callbackSaved(2)}; + KJ_LOG(WARNING, "call through retained capability returned", result); + KJ_EXPECT(result == 1002); + foo->saveCallback(nullptr); +} + +//! FooInterface server that handles callback() like a peer not using the +//! generated code: it keeps the FooCallback capability it was passed and +//! returns without destroying it. +class RetainingFooServer : public ProxyServer +{ +public: + using ProxyServer::ProxyServer; + ~RetainingFooServer() noexcept override = default; + kj::Promise callback(CallbackContext call_context) override + { + m_retained.emplace(call_context.getParams().getCallback()); + call_context.getResults().setResult(0); + return kj::READY_NOW; + } + std::optional m_retained; +}; + +KJ_TEST("repro AR-02: raw peer retaining an Impl& capability after the referent is gone") +{ + class Callback : public FooCallback + { + public: + int call(int arg) override { return arg + m_offset; } + int m_offset{1000}; + }; + + RetainingFooServer* raw_server{nullptr}; + TestSetup setup{/*client_owns_connection=*/true, [&](Connection& connection) -> kj::Own> { + auto server{kj::heap(std::make_shared(), connection)}; + raw_server = server.get(); + return server; + }}; + ProxyClient* foo = setup.client.get(); + foo->initThreadMap(); + { + auto callback{std::make_unique()}; + KJ_EXPECT(foo->callback(*callback, 1) == 0); + } // referent destroyed; the exported ProxyServer still points at it + KJ_REQUIRE(raw_server && raw_server->m_retained.has_value()); + + // The peer now calls the retained capability on a thread it creates + // through the other side's ThreadMap. + EventLoop& loop{*setup.server->m_context.loop}; + std::promise outcome; + loop.sync([&] { + auto make_thread{setup.server->m_context.connection->m_thread_map.makeThreadRequest()}; + make_thread.setName("raw-peer"); + loop.m_task_set->add(make_thread.send().then([&](auto&& results) { + Thread::Client thread{results.getResult()}; + auto call{raw_server->m_retained->callRequest()}; + call.initContext().setThread(thread); + call.setArg(2); + return call.send().then( + [&, thread](auto&& r) { outcome.set_value("retained call returned " + std::to_string(r.getResult())); }, + [&, thread](kj::Exception&& e) { outcome.set_value(std::string{"retained call failed: "} + e.getDescription().cStr()); }); + })); + }); + const std::string result{outcome.get_future().get()}; + KJ_LOG(WARNING, result); + KJ_EXPECT(result.starts_with("retained call failed")); + loop.sync([&] { raw_server->m_retained.reset(); }); +} + +static int ThreadCount() +{ + FILE* f = fopen("/proc/self/status", "r"); + char line[256]; + int threads = -1; + while (f && fgets(line, sizeof(line), f)) { + if (strncmp(line, "Threads:", 8) == 0) threads = atoi(line + 8); + } + if (f) fclose(f); + return threads; +} + +KJ_TEST("repro AR-04: makePool count is not limited") +{ + TestSetup setup; + ProxyClient* foo = setup.client.get(); + foo->initThreadMap(); + const uint32_t count{4000}; + const int before{ThreadCount()}; + const auto start{std::chrono::steady_clock::now()}; + std::promise done; + foo->m_context.loop->sync([&] { + auto request{foo->m_context.connection->m_thread_map.makePoolRequest()}; + request.setCount(count); + foo->m_context.loop->m_task_set->add(request.send().then( + [&](auto&&) { done.set_value("ok"); }, + [&](kj::Exception&& e) { done.set_value(e.getDescription().cStr()); })); + }); + const std::string outcome{done.get_future().get()}; + const auto elapsed{std::chrono::duration_cast(std::chrono::steady_clock::now() - start)}; + const int after{ThreadCount()}; + KJ_LOG(WARNING, "makePool", count, outcome, before, after, elapsed.count()); + KJ_EXPECT(after - before < 1000, "server created one thread per requested pool slot with no cap"); +} + } // namespace test } // namespace mp