From 1153d4f88a2b0d34d1bad5bd90592c706eafead8 Mon Sep 17 00:00:00 2001 From: Markus Werle Date: Fri, 23 Oct 2026 16:54:53 +0200 Subject: [PATCH 1/2] rppgrpc: post the first read before StartCall in client reactors' init() Co-Authored-By: Claude Opus 5.5 --- .../rppgrpc/doxygen/client_reactor.cpp | 6 +-- .../rppgrpc/rppgrpc/client_reactor.hpp | 16 +++++- src/tests/rppgrpc/test_async_client.cpp | 54 +++++++++++++++++++ 3 files changed, 70 insertions(+), 6 deletions(-) diff --git a/src/examples/rppgrpc/doxygen/client_reactor.cpp b/src/examples/rppgrpc/doxygen/client_reactor.cpp index 47ec3ebd6..56e1d29f9 100644 --- a/src/examples/rppgrpc/doxygen/client_reactor.cpp +++ b/src/examples/rppgrpc/doxygen/client_reactor.cpp @@ -19,10 +19,9 @@ int main() // NOLINT(bugprone-exception-escape) const auto reactor = new rppgrpc::client_bidi_reactor(); stub->async()->Bidirectional(&ctx, reactor); reactor->get_observable().subscribe([](const Response&) { }); + reactor->get_observer().on_next(Request{}); // posted until init() starts the call reactor->init(); - - reactor->get_observer().on_next(Request{}); //! [bidi_reactor] } { @@ -49,10 +48,9 @@ int main() // NOLINT(bugprone-exception-escape) Response resp{}; stub->async()->ClientSide(&ctx, &resp, reactor); reactor->get_observable().subscribe([](const rpp::utils::none&) { }); + reactor->get_observer().on_next(Request{}); // posted until init() starts the call reactor->init(); - - reactor->get_observer().on_next(Request{}); //! [write_reactor] } diff --git a/src/extensions/rppgrpc/rppgrpc/client_reactor.hpp b/src/extensions/rppgrpc/rppgrpc/client_reactor.hpp index 2252e2f52..68c4f9924 100644 --- a/src/extensions/rppgrpc/rppgrpc/client_reactor.hpp +++ b/src/extensions/rppgrpc/rppgrpc/client_reactor.hpp @@ -28,6 +28,8 @@ namespace rppgrpc * - to access values FROM stream you can subscribe to observable obtained via `reactor->get_observable()` (same observable WOULD emit on_completed in case of successful stream termination and on_error in case of some errors with grpc stream) * - to pass values TO stream you can emit values to observer obtained via `reactor->get_observer()` * + * @warning After `init()` the call may end at any moment (connection refused, deadline, cancel) and the reactor deletes itself in `OnDone`. Obtain the observer and emit the first values before `init()`: they are posted and sent once the call starts. Emitting from outside the reactor's callbacks after `init()` races with the end of the call (gRPC requires a hold for that, see `grpc::ClientBidiReactor::AddHold`). + * * @snippet client_reactor.cpp bidi_reactor * * @ingroup rppgrpc_reactors @@ -44,8 +46,12 @@ namespace rppgrpc void init() { - Base::StartCall(); + // The first read is posted before StartCall(): gRPC keeps it until the call starts and + // counts it as an outstanding operation. StartCall() is the last use of the reactor in + // init(), because a call that ends at once may run OnDone (and so `delete this`) on a + // gRPC thread before StartCall() has even returned. details::base_reader::handle_read_done(true); + Base::StartCall(); } using details::base_writer::get_observer; @@ -103,6 +109,8 @@ namespace rppgrpc * - to pass values TO stream you can emit values to observer obtained via `reactor->get_observer()` * - reactor provides `reactor->get_observable()` method but such as observable emits nothing and can be used only to be notified about completion/error * + * @warning After `init()` the call may end at any moment (connection refused, deadline, cancel) and the reactor deletes itself in `OnDone`. Obtain the observer and emit the first values before `init()`: they are posted and sent once the call starts. Emitting from outside the reactor's callbacks after `init()` races with the end of the call (gRPC requires a hold for that, see `grpc::ClientWriteReactor::AddHold`). + * * @snippet client_reactor.cpp write_reactor * * @ingroup rppgrpc_reactors @@ -182,8 +190,12 @@ namespace rppgrpc void init() { - Base::StartCall(); + // The first read is posted before StartCall(): gRPC keeps it until the call starts and + // counts it as an outstanding operation. StartCall() is the last use of the reactor in + // init(), because a call that ends at once may run OnDone (and so `delete this`) on a + // gRPC thread before StartCall() has even returned. details::base_reader::handle_read_done(true); + Base::StartCall(); } using details::base_reader::get_observable; diff --git a/src/tests/rppgrpc/test_async_client.cpp b/src/tests/rppgrpc/test_async_client.cpp index e73f7d3cb..4ee18e139 100644 --- a/src/tests/rppgrpc/test_async_client.cpp +++ b/src/tests/rppgrpc/test_async_client.cpp @@ -367,3 +367,57 @@ TEST_CASE("async client reactor") server->Shutdown(); server->Wait(); } + +TEST_CASE("async client reactor: call that ends during init()") +{ + // The context is cancelled before init(), so the call ends as soon as StartCall() is + // issued, and gRPC may run OnDone (which deletes the reactor) on one of its own threads + // while init() has not returned yet. init() must not use the reactor after StartCall(). + // A violation is a use-after-free; AddressSanitizer reports it. + constexpr int iterations = 300; + + grpc::ServerBuilder builder{}; + auto mock_service = std::make_unique(); + ALLOW_CALL(*mock_service, ServerSide(trompeloeil::_, trompeloeil::_, trompeloeil::_)).RETURN(grpc::Status::CANCELLED); + ALLOW_CALL(*mock_service, ClientSide(trompeloeil::_, trompeloeil::_, trompeloeil::_)).RETURN(grpc::Status::CANCELLED); + ALLOW_CALL(*mock_service, Bidirectional(trompeloeil::_, trompeloeil::_)).RETURN(grpc::Status::CANCELLED); + builder.RegisterService(mock_service.get()); + + auto server = builder.BuildAndStart(); + const auto stub = TestService::NewStub(server->InProcessChannel({}), {}); + + // Starts one call with an already cancelled context and waits for the reactor's terminal, + // which it emits from OnDone. Context and request outlive the call. + const auto run_call = [](auto* reactor, auto&& bind) { + grpc::ClientContext ctx{}; + Request request{}; + Response response{}; + const auto done = std::make_shared>(); + reactor->get_observable().subscribe([](const auto&) {}, + [done](const std::exception_ptr&) { done->set_value(); }, + [done] { done->set_value(); }); + bind(ctx, request, response, reactor); + ctx.TryCancel(); + reactor->init(); + return done->get_future().wait_for(std::chrono::seconds{5}) == std::future_status::ready; + }; + + SUBCASE("bidirectional") + { + for (int i = 0; i < iterations; ++i) + REQUIRE(run_call(new rppgrpc::client_bidi_reactor(), [&](auto& ctx, auto&, auto&, auto* r) { stub->async()->Bidirectional(&ctx, r); })); + } + SUBCASE("server-side") + { + for (int i = 0; i < iterations; ++i) + REQUIRE(run_call(new rppgrpc::client_read_reactor(), [&](auto& ctx, auto& req, auto&, auto* r) { stub->async()->ServerSide(&ctx, &req, r); })); + } + SUBCASE("client-side") + { + for (int i = 0; i < iterations; ++i) + REQUIRE(run_call(new rppgrpc::client_write_reactor(), [&](auto& ctx, auto&, auto& resp, auto* r) { stub->async()->ClientSide(&ctx, &resp, r); })); + } + + server->Shutdown(); + server->Wait(); +} From 6a76aab25d598dabd71fded82b2fcfbce60c0530 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:39:30 +0000 Subject: [PATCH 2/2] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- src/tests/rppgrpc/test_async_client.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/tests/rppgrpc/test_async_client.cpp b/src/tests/rppgrpc/test_async_client.cpp index 4ee18e139..e4d0f1bdc 100644 --- a/src/tests/rppgrpc/test_async_client.cpp +++ b/src/tests/rppgrpc/test_async_client.cpp @@ -393,7 +393,7 @@ TEST_CASE("async client reactor: call that ends during init()") Request request{}; Response response{}; const auto done = std::make_shared>(); - reactor->get_observable().subscribe([](const auto&) {}, + reactor->get_observable().subscribe([](const auto&) { }, [done](const std::exception_ptr&) { done->set_value(); }, [done] { done->set_value(); }); bind(ctx, request, response, reactor);