feat grpc: support call interruption in middlewares

commit_hash:31fbc1033e5635a40386d94c94fa9e305c7a8696

Author: kpavlov00

grpc/include/userver/ugrpc/server/impl/call_processor.hpp

@@ -10,8 +10,10 @@
 #include <google/protobuf/message.h>
 #include <grpcpp/server_context.h>
 
+#include <userver/logging/log.hpp>
 #include <userver/server/handlers/exceptions.hpp>
 #include <userver/tracing/in_place_span.hpp>
+#include <userver/utils/fast_scope_guard.hpp>
 #include <userver/utils/impl/internal_tag.hpp>
 
 #include <userver/ugrpc/server/exceptions.hpp>
@@ -125,6 +127,9 @@ class CallProcessor final {
         RunOnCallStart();
 
         bool finished = false;
+        const utils::FastScopeGuard post_finish_hooks_guard([this, &finished]() noexcept {
+            RunOnCallFinish(finished ? std::make_optional(std::move(status_)) : std::nullopt);
+        });
 
         // Don't keep the config snapshot for too long, especially for streaming RPCs.
         state_.config_snapshot.reset();
@@ -138,7 +143,7 @@ class CallProcessor final {
         }
 
         if (!engine::current_task::ShouldCancel() && !responder_.IsInterrupted()) {
-            RunOnCallFinish(response);
+            RunPreFinishHooks(response);
             finished = impl::Finish(responder_, response, status_);
         }
 
@@ -183,7 +188,7 @@ class CallProcessor final {
         }
     }
 
-    void RunOnCallFinish(std::optional<Response>& response) {
+    void RunPreFinishHooks(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) {
@@ -197,8 +202,20 @@ class CallProcessor final {
                 }
             }
 
-            // 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->PreSendStatus(middleware_call_context_, status_); });
+        }
+    }
+
+    void RunOnCallFinish(const std::optional<grpc::Status>& status) {
+        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;
+            try {
+                middleware->OnCallFinish(middleware_call_context_, status);
+            } catch (const std::exception& ex) {
+                LOG_WARNING() << "Error in OnCallFinish: " << ex;
+            }
         }
     }
 

grpc/include/userver/ugrpc/server/middlewares/base.hpp

@@ -3,6 +3,7 @@
 /// @file userver/ugrpc/server/middlewares/base.hpp
 /// @brief @copybrief ugrpc::server::MiddlewareBase
 
+#include <optional>
 #include <string>
 
 #include <google/protobuf/message.h>
@@ -118,14 +119,23 @@ class MiddlewareBase {
     ///  * stream: once per response message, that is, 0, 1, more times per Call (RPC).
     virtual void PreSendMessage(MiddlewareCallContext& context, google::protobuf::Message& response) const;
 
-    /// @brief This hook is invoked once per Call (RPC), after the handler function returns, but before the message is
-    /// sent to the upstream client.
+    /// @brief The function is invoked before sending the final status of the call.
     ///
-    /// All OnCallStart invoked in the reverse order relatively OnCallFinish. You can change grpc status and it will
-    /// apply for a rpc call.
+    /// PreSendStatus is called exactly once per Call (RPC), right before sending
+    /// the final gRPC status to the client. This allows middlewares to inspect
+    /// and potentially modify the status that will be sent to the client.
+    virtual void PreSendStatus(MiddlewareCallContext& context, grpc::Status& status) const;
+
+    /// @brief This hook is invoked once per Call (RPC), after the handler function
+    /// has finished execution and the final status is determined.
     ///
-    /// @warning If Handler (grpc method) returns !ok status, OnCallFinish won't be called.
-    virtual void OnCallFinish(MiddlewareCallContext& context, const grpc::Status& status) const;
+    /// OnCallFinish is called exactly once per Call (RPC), regardless of whether
+    /// the call succeeded or failed. It's the final middleware hook in the call chain.
+    /// This is useful for cleanup operations, logging, or metrics collection that should
+    /// happen after the RPC is completely processed.
+    /// @param context The middleware call context containing call information
+    /// @param status The final status of the call, if available
+    virtual void OnCallFinish(MiddlewareCallContext& context, const std::optional<grpc::Status>& status) const;
 };
 
 /// @ingroup userver_base_classes

grpc/include/userver/ugrpc/server/middlewares/deadline_propagation/middleware.hpp

@@ -25,7 +25,7 @@ class Middleware final : public MiddlewareBase {
 
     void OnCallStart(MiddlewareCallContext& context) const override;
 
-    void OnCallFinish(MiddlewareCallContext& context, const grpc::Status& status) const override;
+    void PreSendStatus(MiddlewareCallContext& context, grpc::Status& status) const override;
 };
 
 }  // namespace ugrpc::server::middlewares::deadline_propagation

grpc/src/ugrpc/server/impl/call_processor.cpp

@@ -1,11 +1,5 @@
 #include <userver/ugrpc/server/impl/call_processor.hpp>
 
-#include <chrono>
-
-#include <boost/container/flat_map.hpp>
-
-#include <userver/logging/impl/logger_base.hpp>
-#include <userver/logging/log.hpp>
 #include <userver/tracing/opentelemetry.hpp>
 #include <userver/tracing/tags.hpp>
 #include <userver/utils/algo.hpp>
@@ -134,11 +128,6 @@ void ReportFinished(const grpc::Status& status, CallState& state) noexcept {
 
 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.
-        LOG_WARNING()
-            << "RPC interrupted in '" << state.call_name
-            << "'. 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::kErrorFlag, true);

grpc/src/ugrpc/server/impl/format_log_message.cpp

@@ -6,11 +6,13 @@
 
 #include <userver/logging/impl/logger_base.hpp>
 #include <userver/logging/impl/timestamp.hpp>
-#include <userver/ugrpc/impl/to_string.hpp>
 #include <userver/utils/datetime.hpp>
 #include <userver/utils/encoding/tskv.hpp>
 #include <userver/utils/text_light.hpp>
 
+#include <userver/ugrpc/impl/to_string.hpp>
+#include <userver/ugrpc/status_codes.hpp>
+
 USERVER_NAMESPACE_BEGIN
 
 namespace ugrpc::server::impl {
@@ -53,7 +55,7 @@ logging::impl::LogExtraTskvFormatter FormatLogMessage(
     std::string_view peer,
     std::chrono::system_clock::time_point start_time,
     std::string_view call_name,
-    grpc::StatusCode code,
+    std::optional<grpc::StatusCode> code,
     const logging::LogExtra* log_extra
 ) {
     static const auto kTimezone = utils::datetime::LocalTimezoneTimestring(start_time, "%z");
@@ -96,8 +98,8 @@ logging::impl::LogExtraTskvFormatter FormatLogMessage(
         // TODO remove, this is for safe migration from old access log parsers.
         request_time_seconds.count(),
         request_time_milliseconds.count(),
-        static_cast<int>(code),
-        ToString(code)
+        code.has_value() ? std::to_string(static_cast<int>(*code)) : "-",
+        code.has_value() ? ToString(*code) : "-"
     );
 
     if (log_extra) {

grpc/src/ugrpc/server/impl/format_log_message.hpp

@@ -1,13 +1,15 @@
 #pragma once
 
-#include <google/protobuf/message.h>
+#include <chrono>
+#include <map>
+#include <optional>
+#include <string_view>
+
+#include <grpcpp/support/status_code_enum.h>
 #include <grpcpp/support/string_ref.h>
 
-#include <userver/dynamic_config/snapshot.hpp>
 #include <userver/logging/impl/log_extra_tskv_formatter.hpp>
-#include <userver/logging/impl/logger_base.hpp>
 #include <userver/logging/log_extra.hpp>
-#include <userver/ugrpc/status_codes.hpp>
 
 USERVER_NAMESPACE_BEGIN
 
@@ -18,7 +20,7 @@ logging::impl::LogExtraTskvFormatter FormatLogMessage(
     std::string_view peer,
     std::chrono::system_clock::time_point start_time,
     std::string_view call_name,
-    grpc::StatusCode code,
+    std::optional<grpc::StatusCode> code,
     const logging::LogExtra* log_extra
 );
 

grpc/src/ugrpc/server/middlewares/access_log/middleware.cpp

@@ -13,7 +13,7 @@ namespace {
 
 void WriteAccessLog(
     MiddlewareCallContext& context,
-    const grpc::Status& status,
+    const std::optional<grpc::Status>& status,
     logging::TextLoggerRef access_tskv_logger
 ) noexcept {
     try {
@@ -30,7 +30,7 @@ void WriteAccessLog(
                     server_context.peer(),
                     context.GetSpan().GetStartSystemTime(),
                     context.GetCallName(),
-                    status.error_code(),
+                    status.has_value() ? std::make_optional(status->error_code()) : std::nullopt,
                     log_extra
                 )
                     .ExtractTextLogItem()
@@ -47,7 +47,7 @@ Middleware::Middleware(Settings&& settings)
     : logger_(std::move(settings.access_tskv_logger))
 {}
 
-void Middleware::OnCallFinish(MiddlewareCallContext& context, const grpc::Status& status) const {
+void Middleware::OnCallFinish(MiddlewareCallContext& context, const std::optional<grpc::Status>& status) const {
     WriteAccessLog(context, status, *logger_);
 }
 

grpc/src/ugrpc/server/middlewares/access_log/middleware.hpp

@@ -17,7 +17,7 @@ class Middleware final : public MiddlewareBase {
 public:
     explicit Middleware(Settings&& settings);
 
-    void OnCallFinish(MiddlewareCallContext& context, const grpc::Status& status) const override;
+    void OnCallFinish(MiddlewareCallContext& context, const std::optional<grpc::Status>& status) const override;
 
 private:
     logging::TextLoggerPtr logger_;

grpc/src/ugrpc/server/middlewares/base.cpp

@@ -51,7 +51,9 @@ void MiddlewareBase::PostRecvMessage(MiddlewareCallContext&, google::protobuf::M
 
 void MiddlewareBase::PreSendMessage(MiddlewareCallContext&, google::protobuf::Message&) const {}
 
-void MiddlewareBase::OnCallFinish(MiddlewareCallContext&, const grpc::Status&) const {}
+void MiddlewareBase::PreSendStatus(MiddlewareCallContext&, grpc::Status&) const {}
+
+void MiddlewareBase::OnCallFinish(MiddlewareCallContext&, const std::optional<grpc::Status>&) const {}
 
 }  // namespace ugrpc::server
 

grpc/src/ugrpc/server/middlewares/deadline_propagation/middleware.cpp

@@ -91,7 +91,7 @@ void Middleware::OnCallStart(MiddlewareCallContext& context) const {
     }
 }
 
-void Middleware::OnCallFinish(MiddlewareCallContext& context, const grpc::Status& status) const {
+void Middleware::PreSendStatus(MiddlewareCallContext& context, grpc::Status& status) const {
     const auto* const inherited_data = USERVER_NAMESPACE::server::request::kTaskInheritedData.GetOptional();
 
     // if !USERVER_DEADLINE_PROPAGATION_ENABLED, inherited_data must be nullptr