refactor grpc: refactor server::impl::CallProcessor

kpavlov00 · kpavlov00 · commit 7dde47ed45ae · 2025-12-05T00:00:08.000+03:00
commit_hash:f0bd4b271eaee251f31980cf36da1fd376870523
diff --git a/grpc/include/userver/ugrpc/client/impl/unary_call.hpp b/grpc/include/userver/ugrpc/client/impl/unary_call.hpp
@@ -7,6 +7,7 @@
 #include <userver/engine/sleep.hpp>
 #include <userver/server/request/task_inherited_data.hpp>
 #include <userver/tracing/tags.hpp>
+#include <userver/utils/fast_scope_guard.hpp>
 #include <userver/utils/impl/internal_tag.hpp>
 
 #include <userver/ugrpc/client/call_context.hpp>
diff --git a/grpc/include/userver/ugrpc/server/impl/call_processor.hpp b/grpc/include/userver/ugrpc/server/impl/call_processor.hpp
@@ -35,14 +35,14 @@ void SetupSpan(
     std::string_view method_name
 );
 
-grpc::Status ReportHandlerError(const std::exception& ex, CallState& state) noexcept;
-
-void ReportRpcInterruptedError(CallState& state) noexcept;
-
 grpc::Status ReportCustomError(const USERVER_NAMESPACE::server::handlers::CustomHandlerException& ex, CallState& state)
     noexcept;
 
-void ReportFinish(bool finish_op_succeeded, const grpc::Status& status, CallState& state) noexcept;
+grpc::Status ReportHandlerError(const std::exception& ex, CallState& state) noexcept;
+
+void ReportFinished(const grpc::Status& status, CallState& state) noexcept;
+
+void ReportInterrupted(CallState& state) noexcept;
 
 template <typename Response>
 void UnpackResult(Result<Response>&& result, std::optional<Response>& response, grpc::Status& status) {
@@ -64,6 +64,28 @@ void UnpackResult(StreamingResult<Response>&& result, std::optional<Response>& r
     }
 }
 
+template <typename CallTraits>
+bool Finish(
+    impl::Responder<CallTraits>& responder,
+    const std::optional<typename CallTraits::Response>& response,
+    grpc::Status& status
+) {
+    if (status.ok()) {
+        if constexpr (IsServerStreaming(CallTraits::kCallKind)) {
+            if (response.has_value()) {
+                return responder.Finish(*response);
+            } else {
+                return responder.Finish();
+            }
+        } else {
+            UINVARIANT(response.has_value(), "response should not be empty");
+            return responder.Finish(*response);
+        }
+    } else {
+        return responder.FinishWithError(status);
+    }
+}
+
 template <typename CallTraits>
 class CallProcessor final {
 public:
@@ -84,7 +106,7 @@ class CallProcessor final {
     )
         : state_(std::move(params), CallTraits::kCallKind),
           responder_(state_, raw_responder),
-          middleware_call_context_(utils::impl::InternalTag{}, state_),
+          middleware_call_context_(utils::impl::InternalTag{}, state_, status_),
           initial_request_(initial_request),
           service_(service),
           service_method_(service_method)
@@ -102,52 +124,29 @@ class CallProcessor final {
     void DoCall() {
         RunOnCallStart();
 
+        bool finished = false;
+
         // Don't keep the config snapshot for too long, especially for streaming RPCs.
         state_.config_snapshot.reset();
 
-        // Final response is the response sent to the client together with status in the final batch.
-        std::optional<Response> final_response;
-
-        if (!Status().ok()) {
-            RunOnCallFinish(final_response);
-            impl::ReportFinish(responder_.FinishWithError(Status()), Status(), state_);
-            return;
+        std::optional<Response> response;
+        if (!engine::current_task::ShouldCancel() && status_.ok()) {
+            RunWithCatch([this, &response] {
+                auto result = CallHandler();
+                impl::UnpackResult(std::move(result), response, status_);
+            });
         }
 
-        RunWithCatch([this, &final_response] {
-            auto result = CallHandler();
-            impl::UnpackResult(std::move(result), final_response, Status());
-        });
-
-        // Streaming handler can detect RPC breakage during a network interaction => IsInterrupted().
-        // RpcFinishedEvent can signal RPC interruption while in the handler => ShouldCancel.
-        if (responder_.IsInterrupted() || engine::current_task::ShouldCancel()) {
-            impl::ReportRpcInterruptedError(state_);
-            // Don't run OnCallFinish.
-            return;
+        if (!engine::current_task::ShouldCancel() && !responder_.IsInterrupted()) {
+            RunOnCallFinish(response);
+            finished = impl::Finish(responder_, response, status_);
         }
 
-        if (!Status().ok()) {
-            RunOnCallFinish(final_response);
-            impl::ReportFinish(responder_.FinishWithError(Status()), Status(), state_);
-            return;
-        }
-
-        RunOnCallFinish(final_response);
-
-        if (!Status().ok()) {
-            impl::ReportFinish(responder_.FinishWithError(Status()), Status(), state_);
-            return;
-        }
-
-        if constexpr (IsServerStreaming(CallTraits::kCallKind)) {
-            if (!final_response) {
-                impl::ReportFinish(responder_.Finish(), Status(), state_);
-                return;
-            }
+        if (finished) {
+            impl::ReportFinished(status_, state_);
+        } else {
+            impl::ReportInterrupted(state_);
         }
-        UASSERT(final_response);
-        impl::ReportFinish(responder_.Finish(*final_response), Status(), state_);
     }
 
 private:
@@ -169,37 +168,37 @@ class CallProcessor final {
         UASSERT(success_pre_hooks_count_ == 0);
         for (const auto& m : state_.middlewares) {
             RunWithCatch([this, &m] { m->OnCallStart(middleware_call_context_); });
-            if (!Status().ok()) {
+            if (!status_.ok()) {
                 return;
             }
             // On fail, we must call OnRpcFinish only for middlewares for which OnRpcStart has been called successfully.
             // So, we watch to count of these middlewares.
             ++success_pre_hooks_count_;
             if constexpr (std::is_base_of_v<google::protobuf::Message, InitialRequest>) {
                 RunWithCatch([this, &m] { m->PostRecvMessage(middleware_call_context_, initial_request_); });
-                if (!Status().ok()) {
+                if (!status_.ok()) {
                     return;
                 }
             }
         }
     }
 
-    void RunOnCallFinish(std::optional<Response>& final_response) {
+    void RunOnCallFinish(std::optional<Response>& response) {
         const auto& mids = state_.middlewares;
         const auto rbegin = mids.rbegin() + (mids.size() - success_pre_hooks_count_);
         for (auto it = rbegin; it != mids.rend(); ++it) {
             const auto& middleware = *it;
 
             if constexpr (std::is_base_of_v<google::protobuf::Message, Response>) {
-                if (Status().ok() && final_response.has_value()) {
-                    RunWithCatch([this, &middleware, &final_response] {
-                        middleware->PreSendMessage(middleware_call_context_, *final_response);
+                if (status_.ok() && response.has_value()) {
+                    RunWithCatch([this, &middleware, &response] {
+                        middleware->PreSendMessage(middleware_call_context_, *response);
                     });
                 }
             }
 
             // We must call all OnRpcFinish despite the failures. So, don't check the status.
-            RunWithCatch([this, &middleware] { middleware->OnCallFinish(middleware_call_context_, Status()); });
+            RunWithCatch([this, &middleware] { middleware->OnCallFinish(middleware_call_context_, status_); });
         }
     }
 
@@ -208,23 +207,22 @@ class CallProcessor final {
         try {
             func();
         } catch (MiddlewareRpcInterruptionError& ex) {
-            Status() = ex.ExtractStatus();
+            status_ = ex.ExtractStatus();
         } catch (ErrorWithStatus& ex) {
-            Status() = ex.ExtractStatus();
+            status_ = ex.ExtractStatus();
         } catch (const USERVER_NAMESPACE::server::handlers::CustomHandlerException& ex) {
-            Status() = impl::ReportCustomError(ex, state_);
+            status_ = impl::ReportCustomError(ex, state_);
         } catch (const RpcInterruptedError& /*ex*/) {
             UASSERT(responder_.IsInterrupted());
             // RPC interruption will be reported below.
         } catch (const std::exception& ex) {
-            Status() = impl::ReportHandlerError(ex, state_);
+            status_ = impl::ReportHandlerError(ex, state_);
         }
     }
 
-    grpc::Status& Status() { return middleware_call_context_.GetStatus(utils::impl::InternalTag{}); }
-
     CallState state_;
     Responder responder_;
+    grpc::Status status_;
     MiddlewareCallContext middleware_call_context_;
     // Initial request is the request which is sent to the service together with RPC initiation.
     // Unary-request RPCs have an initial request, client-streaming RPCs don't.
diff --git a/grpc/include/userver/ugrpc/server/impl/call_state.hpp b/grpc/include/userver/ugrpc/server/impl/call_state.hpp
@@ -3,6 +3,8 @@
 #include <optional>
 #include <string_view>
 
+#include <boost/container/flat_map.hpp>
+
 #include <grpcpp/completion_queue.h>
 #include <grpcpp/server_context.h>
 
diff --git a/grpc/include/userver/ugrpc/server/middlewares/base.hpp b/grpc/include/userver/ugrpc/server/middlewares/base.hpp
@@ -43,7 +43,7 @@ class MiddlewareCallContext final : public CallContextBase {
 public:
     /// @cond
     // For internal use only
-    MiddlewareCallContext(utils::impl::InternalTag, impl::CallState& state);
+    MiddlewareCallContext(utils::impl::InternalTag, impl::CallState& state, grpc::Status& status);
     /// @endcond
 
     /// @brief Aborts the RPC, returning the specified status to the upstream client, see details below.
@@ -80,13 +80,10 @@ class MiddlewareCallContext final : public CallContextBase {
     /// @cond
     // For internal use only.
     ugrpc::impl::RpcStatisticsScope& GetStatistics(utils::impl::InternalTag);
-
-    // For internal use only.
-    grpc::Status& GetStatus(utils::impl::InternalTag) { return status_; }
     /// @endcond
 
 private:
-    grpc::Status status_;
+    grpc::Status& status_;
 };
 
 /// @ingroup userver_base_classes userver_grpc_server_middlewares
diff --git a/grpc/src/ugrpc/server/impl/call_processor.cpp b/grpc/src/ugrpc/server/impl/call_processor.cpp
@@ -22,26 +22,6 @@ namespace ugrpc::server::impl {
 
 namespace {
 
-void ReportFinishSuccess(const grpc::Status& status, CallState& state) noexcept {
-    try {
-        state.statistics_scope.OnExplicitFinish(status.error_code());
-
-        auto& span = state.GetSpan();
-        span.AddNonInheritableTag(tracing::kGrpcCode, ugrpc::ToString(status.error_code()));
-        if (!status.ok()) {
-            span.AddNonInheritableTag(tracing::kErrorFlag, true);
-            span.AddNonInheritableTag(tracing::kErrorMessage, status.error_message());
-            const auto default_error_log_level =
-                IsServerError(status.error_code()) ? logging::Level::kError : logging::Level::kWarning;
-            const auto error_log_level =
-                utils::FindOrDefault(state.status_codes_log_level, status.error_code(), default_error_log_level);
-            span.SetLogLevel(error_log_level);
-        }
-    } catch (const std::exception& ex) {
-        LOG_ERROR() << "Error in ReportFinishSuccess: " << ex;
-    }
-}
-
 logging::Level AdjustLogLevelForCancellations(logging::Level level) {
     return engine::current_task::ShouldCancel() ? std::min(level, logging::Level::kWarning) : level;
 }
@@ -98,6 +78,26 @@ void SetupSpan(
     span.AddNonInheritableTag(tracing::kRpcMethod, std::string{method_name});
 }
 
+grpc::Status ReportCustomError(const USERVER_NAMESPACE::server::handlers::CustomHandlerException& ex, CallState& state)
+    noexcept {
+    try {
+        grpc::Status status{CustomStatusToGrpc(ex.GetCode()), ex.GetExternalErrorBody()};
+
+        const auto log_level = AdjustLogLevelForCancellations(
+            IsServerError(status.error_code()) ? logging::Level::kError : logging::Level::kWarning
+        );
+        LOG(log_level) << "Error in " << state.call_name << ": " << ex;
+        auto& span = state.GetSpan();
+        span.AddNonInheritableTag(tracing::kErrorFlag, true);
+        span.AddNonInheritableTag(tracing::kErrorMessage, ex.what());
+        span.SetLogLevel(log_level);
+        return status;
+    } catch (const std::exception& new_ex) {
+        LOG_ERROR() << "Error in ReportCustomError: " << new_ex;
+        return grpc::Status{grpc::StatusCode::INTERNAL, ""};
+    }
+}
+
 grpc::Status ReportHandlerError(const std::exception& ex, CallState& state) noexcept {
     try {
         auto& span = state.GetSpan();
@@ -113,7 +113,26 @@ grpc::Status ReportHandlerError(const std::exception& ex, CallState& state) noex
     }
 }
 
-void ReportRpcInterruptedError(CallState& state) noexcept {
+void ReportFinished(const grpc::Status& status, CallState& state) noexcept {
+    try {
+        state.statistics_scope.OnExplicitFinish(status.error_code());
+        auto& span = state.GetSpan();
+        span.AddNonInheritableTag(tracing::kGrpcCode, ugrpc::ToString(status.error_code()));
+        if (!status.ok()) {
+            span.AddNonInheritableTag(tracing::kErrorFlag, true);
+            span.AddNonInheritableTag(tracing::kErrorMessage, status.error_message());
+            const auto default_error_log_level =
+                IsServerError(status.error_code()) ? logging::Level::kError : logging::Level::kWarning;
+            const auto error_log_level =
+                utils::FindOrDefault(state.status_codes_log_level, status.error_code(), default_error_log_level);
+            span.SetLogLevel(error_log_level);
+        }
+    } catch (const std::exception& ex) {
+        LOG_ERROR() << "Error in ReportFinished: " << ex;
+    }
+}
+
+void ReportInterrupted(CallState& state) noexcept {
     try {
         // RPC interruption leads to asynchronous task cancellation by RpcFinishedEvent,
         // so the task either is already cancelled, or is going to be cancelled.
@@ -122,39 +141,11 @@ void ReportRpcInterruptedError(CallState& state) noexcept {
             << "'. The previously logged cancellation or network exception, if any, is likely caused by it.";
         state.statistics_scope.OnNetworkError();
         auto& span = state.GetSpan();
-        span.AddNonInheritableTag(tracing::kErrorMessage, "RPC interrupted");
         span.AddNonInheritableTag(tracing::kErrorFlag, true);
+        span.AddNonInheritableTag(tracing::kErrorMessage, "RPC interrupted");
         span.SetLogLevel(logging::Level::kWarning);
     } catch (const std::exception& ex) {
-        LOG_ERROR() << "Error in ReportRpcInterruptedError: " << ex;
-    }
-}
-
-grpc::Status ReportCustomError(const USERVER_NAMESPACE::server::handlers::CustomHandlerException& ex, CallState& state)
-    noexcept {
-    try {
-        grpc::Status status{CustomStatusToGrpc(ex.GetCode()), ex.GetExternalErrorBody()};
-
-        const auto log_level = AdjustLogLevelForCancellations(
-            IsServerError(status.error_code()) ? logging::Level::kError : logging::Level::kWarning
-        );
-        LOG(log_level) << "Error in " << state.call_name << ": " << ex;
-        auto& span = state.GetSpan();
-        span.AddNonInheritableTag(tracing::kErrorFlag, true);
-        span.AddNonInheritableTag(tracing::kErrorMessage, ex.what());
-        span.SetLogLevel(log_level);
-        return status;
-    } catch (const std::exception& new_ex) {
-        LOG_ERROR() << "Error in ReportCustomError: " << new_ex;
-        return grpc::Status{grpc::StatusCode::INTERNAL, ""};
-    }
-}
-
-void ReportFinish(bool finish_op_succeeded, const grpc::Status& status, CallState& state) noexcept {
-    if (finish_op_succeeded) {
-        ReportFinishSuccess(status, state);
-    } else {
-        ReportRpcInterruptedError(state);
+        LOG_ERROR() << "Error in ReportInterrupted: " << ex;
     }
 }
 
diff --git a/grpc/src/ugrpc/server/impl/rpc.cpp b/grpc/src/ugrpc/server/impl/rpc.cpp
@@ -31,11 +31,10 @@ std::unique_lock<engine::SingleWaitingTaskMutex> ResponderBase::TakeMutexIfBidir
 
 void ResponderBase::ApplyRequestHook(google::protobuf::Message& request) {
     auto lock = TakeMutexIfBidirectional();
-    MiddlewareCallContext middleware_call_context{utils::impl::InternalTag{}, state_};
-
+    grpc::Status status;
+    MiddlewareCallContext middleware_call_context{utils::impl::InternalTag{}, state_, status};
     for (const auto& middleware : state_.middlewares) {
         middleware->PostRecvMessage(middleware_call_context, request);
-        auto& status = middleware_call_context.GetStatus(utils::impl::InternalTag{});
         if (!status.ok()) {
             throw impl::MiddlewareRpcInterruptionError(std::move(status));
         }
@@ -44,11 +43,10 @@ void ResponderBase::ApplyRequestHook(google::protobuf::Message& request) {
 
 void ResponderBase::ApplyResponseHook(google::protobuf::Message& response) {
     auto lock = TakeMutexIfBidirectional();
-    MiddlewareCallContext middleware_call_context{utils::impl::InternalTag{}, state_};
-
+    grpc::Status status;
+    MiddlewareCallContext middleware_call_context{utils::impl::InternalTag{}, state_, status};
     for (const auto& middleware : boost::adaptors::reverse(state_.middlewares)) {
         middleware->PreSendMessage(middleware_call_context, response);
-        auto& status = middleware_call_context.GetStatus(utils::impl::InternalTag{});
         if (!status.ok()) {
             throw impl::MiddlewareRpcInterruptionError(std::move(status));
         }
diff --git a/grpc/src/ugrpc/server/middlewares/base.cpp b/grpc/src/ugrpc/server/middlewares/base.cpp
diff --git a/grpc/src/ugrpc/server/middlewares/log/middleware.hpp b/grpc/src/ugrpc/server/middlewares/log/middleware.hpp
diff --git a/grpc/tests/tests/server_middleware_base_gmock.hpp b/grpc/tests/tests/server_middleware_base_gmock.hpp
