rppgrpc: client reactors post the first read before StartCall

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<Request, Response>();
         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]
     }
 

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<Response>::handle_read_done(true);
+            Base::StartCall();
         }
 
         using details::base_writer<Request>::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<Response>::handle_read_done(true);
+            Base::StartCall();
         }
 
         using details::base_reader<Response>::get_observable;

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<service>();
+    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<std::promise<void>>();
+        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<Request, Response>(), [&](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<Response>(), [&](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<Request>(), [&](auto& ctx, auto&, auto& resp, auto* r) { stub->async()->ClientSide(&ctx, &resp, r); }));
+    }
+
+    server->Shutdown();
+    server->Wait();
+}