fix grpc: coredump from ReadRemainingAndFinish & remove PingPongFinish

Tests: CI
commit_hash:aaf5e31181ffc08d43627972e73ec9761a36b1a9

Author: TTPO100AJIEX

grpc/include/userver/ugrpc/client/graceful_stream_finish.hpp

@@ -26,7 +26,12 @@ namespace ugrpc::client {
 ///         or the server returning an error status)
 template <typename Response>
 std::optional<std::size_t> ReadRemainingAndFinish(Reader<Response>& stream) noexcept {
-    return impl::ReadRemainingAndFinish<Reader<Response>, Response>(stream);
+    try {
+        return impl::ReadRemainingAndFinish<Reader<Response>, Response>(stream);
+    } catch (const std::exception& e) {
+        LOG_WARNING() << "Failed to read remaining messages and finish - " << e;
+        return std::nullopt;
+    }
 }
 
 /// @brief Announce end-of-input to the server, read all
@@ -41,40 +46,14 @@ std::optional<std::size_t> ReadRemainingAndFinish(Reader<Response>& stream) noex
 ///         either reads or writes, or the server returning an error status)
 template <typename Request, typename Response>
 std::optional<std::size_t> ReadRemainingAndFinish(ReaderWriter<Request, Response>& stream) noexcept {
-    const bool writes_done_success = stream.WritesDone();
-    const std::optional<std::size_t>
-        messages_remaining = impl::ReadRemainingAndFinish<ReaderWriter<Request, Response>, Response>(stream);
-    return writes_done_success ? messages_remaining : std::nullopt;
-}
-
-/// @brief Gracefully finish a ping-pong style interaction
-///
-/// 1. Announces end-of-output to the server
-/// 2. Ensures there are no more messages to read
-///
-/// @warning The method will hang indefenitely if the stream
-///          is never interrupted or closed by the server
-///
-/// @return true if the operation was successful; false if there are more messages
-///         in the stream to read or if the operation failed (i.e. due to task
-///         cancellation, the stream being already closed for writes,
-///         or the server returning an error status)
-template <typename Request, typename Response>
-[[nodiscard]] bool PingPongFinish(ReaderWriter<Request, Response>& stream) noexcept {
     try {
         const bool writes_done_success = stream.WritesDone();
-
-        Response response;
-        if (!stream.Read(response)) {
-            // As expected, there are no more messages in the stream
-            return writes_done_success;
-        }
-
-        LOG_WARNING() << "PingPongFinish: there are more messages to read from the stream";
-        return false;
+        const std::optional<std::size_t>
+            messages_remaining = impl::ReadRemainingAndFinish<ReaderWriter<Request, Response>, Response>(stream);
+        return writes_done_success ? messages_remaining : std::nullopt;
     } catch (const std::exception& e) {
-        LOG_WARNING() << "PingPongFinish: failed to close the stream - " << e;
-        return false;
+        LOG_WARNING() << "Failed to read remaining messages and finish - " << e;
+        return std::nullopt;
     }
 }
 

grpc/include/userver/ugrpc/client/impl/graceful_stream_finish.hpp

@@ -11,18 +11,13 @@ USERVER_NAMESPACE_BEGIN
 namespace ugrpc::client::impl {
 
 template <typename Stream, typename Response>
-std::optional<std::size_t> ReadRemainingAndFinish(Stream& stream) noexcept {
-    try {
-        Response response;
-        std::size_t num_messages = 0;
-        while (stream.Read(response)) {
-            ++num_messages;
-        }
-        return num_messages;
-    } catch (const std::exception& e) {
-        LOG_WARNING() << "Failed to read remaining messages and finish - " << e;
-        return std::nullopt;
+std::optional<std::size_t> ReadRemainingAndFinish(Stream& stream) {
+    Response response;
+    std::size_t num_messages = 0;
+    while (stream.Read(response)) {
+        ++num_messages;
     }
+    return num_messages;
 }
 
 }  // namespace ugrpc::client::impl

grpc/include/userver/ugrpc/client/stream.hpp

@@ -146,7 +146,7 @@ class [[nodiscard]] Writer final {
 /// If destroyed early, the RPC is cancelled. The server gets @ref ugrpc::client::RpcInterruptedError
 /// and the `abandoned-error` metric is incremented. The connection stays open for reuse.
 /// gRPC provides no way to early-close a server-streaming RPC gracefully.
-// See @ref ugrpc::client::ReadRemainingAndFinish and @ref ugrpc::client::PingPongFinish for graceful completion.
+// See @ref ugrpc::client::ReadRemainingAndFinish for graceful completion.
 ///
 /// `Read` and `AsyncRead` can throw if error status is received from server.
 /// User MUST NOT call `Read` or `AsyncRead` again after failure of any of these

grpc/tests/stream_test.cpp

@@ -107,66 +107,6 @@ UTEST_F_MT(GrpcBidirectionalStream, BidirectionalStreamTest, 2) {
     );
 }
 
-UTEST_F(GrpcBidirectionalStream, PingPongFinishOk) {
-    auto stream = GetClient().Chat();
-
-    ASSERT_TRUE(stream.Write(sample::ugrpc::StreamGreetingRequest()));
-    sample::ugrpc::StreamGreetingResponse response;
-    ASSERT_TRUE(stream.Read(response));
-
-    ASSERT_TRUE(ugrpc::client::PingPongFinish(stream));
-
-    ASSERT_FALSE(stream.Write(sample::ugrpc::StreamGreetingRequest()));
-    ASSERT_FALSE(stream.WritesDone());
-    ASSERT_THROW(stream.WriteAndCheck(sample::ugrpc::StreamGreetingRequest()), ugrpc::client::RpcError);
-    ASSERT_FALSE(stream.Read(response));
-    ASSERT_THROW([[maybe_unused]] auto _ = stream.ReadAsync(response), ugrpc::client::RpcError);
-}
-
-UTEST_F(GrpcBidirectionalStream, PingPongFinishNoMessages) {
-    auto stream = GetClient().Chat();
-
-    ASSERT_TRUE(ugrpc::client::PingPongFinish(stream));
-
-    ASSERT_FALSE(stream.Write(sample::ugrpc::StreamGreetingRequest()));
-    ASSERT_FALSE(stream.WritesDone());
-    ASSERT_THROW(stream.WriteAndCheck(sample::ugrpc::StreamGreetingRequest()), ugrpc::client::RpcError);
-    sample::ugrpc::StreamGreetingResponse response;
-    ASSERT_FALSE(stream.Read(response));
-    ASSERT_THROW([[maybe_unused]] auto _ = stream.ReadAsync(response), ugrpc::client::RpcError);
-}
-
-UTEST_F(GrpcBidirectionalStream, PingPongFinishMoreMessages) {
-    auto stream = GetClient().Chat();
-
-    ASSERT_TRUE(stream.Write(sample::ugrpc::StreamGreetingRequest()));
-    // No 'Read' here
-
-    ASSERT_FALSE(ugrpc::client::PingPongFinish(stream));
-
-    ASSERT_FALSE(stream.Write(sample::ugrpc::StreamGreetingRequest()));
-    ASSERT_FALSE(stream.WritesDone());
-    ASSERT_THROW(stream.WriteAndCheck(sample::ugrpc::StreamGreetingRequest()), ugrpc::client::RpcError);
-    sample::ugrpc::StreamGreetingResponse response;
-    ASSERT_FALSE(stream.Read(response));
-    ASSERT_THROW([[maybe_unused]] auto _ = stream.ReadAsync(response), ugrpc::client::RpcError);
-}
-
-UTEST_F(GrpcBidirectionalStream, PingPongFinishAfterWritesDone) {
-    auto stream = GetClient().Chat();
-
-    ASSERT_TRUE(stream.WritesDone());
-
-    ASSERT_FALSE(ugrpc::client::PingPongFinish(stream));
-
-    ASSERT_FALSE(stream.Write(sample::ugrpc::StreamGreetingRequest()));
-    ASSERT_FALSE(stream.WritesDone());
-    ASSERT_THROW(stream.WriteAndCheck(sample::ugrpc::StreamGreetingRequest()), ugrpc::client::RpcError);
-    sample::ugrpc::StreamGreetingResponse response;
-    ASSERT_FALSE(stream.Read(response));
-    ASSERT_THROW([[maybe_unused]] auto _ = stream.ReadAsync(response), ugrpc::client::RpcError);
-}
-
 UTEST_F(GrpcBidirectionalStream, BidirectionalStreamReadRemaining) {
     auto stream = GetClient().Chat();
 
@@ -240,6 +180,31 @@ UTEST_F(GrpcBidirectionalStream, BidirectionalStreamDestroy) {
     EXPECT_EQ(get_metric(kStatus, {{"grpc_code", "UNKNOWN"}}), 0);
 }
 
+namespace {
+
+class UnitTestServiceChatAutoFinish final : public sample::ugrpc::UnitTestServiceBase {
+public:
+    ChatResult Chat(CallContext& /*context*/, ChatReaderWriter& stream) override {
+        stream.Write(sample::ugrpc::StreamGreetingResponse{});
+        return grpc::Status::OK;
+    }
+};
+
+using GrpcBidirectionalStreamChatAutoFinish =
+    ugrpc::tests::ServiceWithClientFixture<UnitTestServiceChatAutoFinish, sample::ugrpc::UnitTestServiceClient>;
+
+}  // namespace
+
+UTEST_F(GrpcBidirectionalStreamChatAutoFinish, BidirectionalStreamWritesDoneThrowOnClosedStream) {
+    auto stream = GetClient().Chat();
+
+    sample::ugrpc::StreamGreetingResponse response;
+    ASSERT_TRUE(stream.Read(response));
+    ASSERT_FALSE(stream.Read(response));
+
+    UASSERT_NO_THROW(ugrpc::client::ReadRemainingAndFinish(stream));
+}
+
 UTEST_F(GrpcInputStream, InputStreamDestroy) {
     {
         const sample::ugrpc::StreamGreetingRequest request;