fix grpc: fix client outbound middleware run order

kpavlov00 · kpavlov00 · commit b7f382f36a33 · 2025-12-10T13:14:33.000+03:00
commit_hash:c24fdd3c5dab22e80aaaabc07f7fc023615c6c71
diff --git a/grpc/include/userver/ugrpc/client/impl/middleware_hooks.hpp b/grpc/include/userver/ugrpc/client/impl/middleware_hooks.hpp
@@ -1,5 +1,7 @@
 #pragma once
 
+#include <variant>
+
 #include <google/protobuf/message.h>
 #include <grpcpp/support/status.h>
 
@@ -14,24 +16,36 @@ namespace ugrpc::client::impl {
 
 class MiddlewareHooks {
 public:
-    void SetStartCall() noexcept;
-    void SetSendMessage(const google::protobuf::Message& send_message) noexcept;
-    void SetRecvMessage(const google::protobuf::Message& recv_message) noexcept;
-    void SetStatus(const grpc::Status& status) noexcept;
+    static MiddlewareHooks StartCallHooks(const google::protobuf::Message* request = nullptr) noexcept;
+
+    static MiddlewareHooks SendMessageHooks(const google::protobuf::Message& send_message) noexcept;
+
+    static MiddlewareHooks RecvMessageHooks(const google::protobuf::Message& recv_message) noexcept;
+
+    static MiddlewareHooks FinishHooks(const grpc::Status& status, const google::protobuf::Message* response = nullptr)
+        noexcept;
 
     void Run(const MiddlewareBase& middleware, MiddlewareCallContext& context) const;
 
+    bool Reverse() const noexcept;
+
 private:
-    bool start_call_{false};
-    const google::protobuf::Message* send_message_{};
-    const google::protobuf::Message* recv_message_{};
-    const grpc::Status* status_{};
-};
+    struct Inbound {
+        bool start_call{false};
+        const google::protobuf::Message* send_message{};
+    };
+
+    struct Outbound {
+        const grpc::Status* status{};
+        const google::protobuf::Message* recv_message{};
+    };
+
+    using Params = std::variant<Inbound, Outbound>;
 
-MiddlewareHooks StartCallHooks(const google::protobuf::Message* request = nullptr) noexcept;
-MiddlewareHooks SendMessageHooks(const google::protobuf::Message& send_message) noexcept;
-MiddlewareHooks RecvMessageHooks(const google::protobuf::Message& recv_message) noexcept;
-MiddlewareHooks FinishHooks(const grpc::Status& status, const google::protobuf::Message* response = nullptr) noexcept;
+    explicit MiddlewareHooks(Params&& params) noexcept;
+
+    Params params_;
+};
 
 }  // namespace ugrpc::client::impl
 
diff --git a/grpc/include/userver/ugrpc/client/impl/rpc.hpp b/grpc/include/userver/ugrpc/client/impl/rpc.hpp
@@ -257,7 +257,7 @@ InputStream<Response>::InputStream(
     : state_{std::move(params), CallKind::kInputStream},
       context_{utils::impl::InternalTag{}, state_}
 {
-    RunMiddlewarePipeline(state_, StartCallHooks(ToBaseMessage(&request)));
+    RunMiddlewarePipeline(state_, MiddlewareHooks::StartCallHooks(ToBaseMessage(&request)));
 
     // NOLINTNEXTLINE(cppcoreguidelines-prefer-member-initializer)
     stream_ = impl::PrepareCall(
@@ -286,7 +286,7 @@ bool InputStream<Response>::Read(Response& response) {
     }
 
     if (impl::Read(*stream_, response, state_)) {
-        RunMiddlewarePipeline(state_, RecvMessageHooks(response));
+        RunMiddlewarePipeline(state_, MiddlewareHooks::RecvMessageHooks(response));
         return true;
     } else {
         // Finish can only be called once all the data is read, otherwise the
@@ -305,7 +305,7 @@ OutputStream<Request, Response>::OutputStream(
     : state_{std::move(params), CallKind::kOutputStream},
       context_{utils::impl::InternalTag{}, state_}
 {
-    RunMiddlewarePipeline(state_, StartCallHooks());
+    RunMiddlewarePipeline(state_, MiddlewareHooks::StartCallHooks());
 
     // 'response_' will be filled upon successful 'Finish' async call
     // NOLINTNEXTLINE(cppcoreguidelines-prefer-member-initializer)
@@ -332,7 +332,7 @@ bool OutputStream<Request, Response>::Write(const Request& request) {
         return false;
     }
 
-    RunMiddlewarePipeline(state_, SendMessageHooks(request));
+    RunMiddlewarePipeline(state_, MiddlewareHooks::SendMessageHooks(request));
 
     // Don't buffer writes, otherwise in an event subscription scenario, events
     // may never actually be delivered
@@ -348,7 +348,7 @@ void OutputStream<Request, Response>::WriteAndCheck(const Request& request) {
         throw RpcError(state_.GetCallName(), "'WriteAndCheck' called on a finished or closed stream");
     }
 
-    RunMiddlewarePipeline(state_, SendMessageHooks(request));
+    RunMiddlewarePipeline(state_, MiddlewareHooks::SendMessageHooks(request));
 
     // Don't buffer writes, otherwise in an event subscription scenario, events
     // may never actually be delivered
@@ -381,7 +381,7 @@ BidirectionalStream<Request, Response>::BidirectionalStream(
     : state_{std::move(params), CallKind::kBidirectionalStream},
       context_{utils::impl::InternalTag{}, state_}
 {
-    RunMiddlewarePipeline(state_, StartCallHooks());
+    RunMiddlewarePipeline(state_, MiddlewareHooks::StartCallHooks());
 
     // NOLINTNEXTLINE(cppcoreguidelines-prefer-member-initializer)
     stream_ = impl::PrepareCall(prepare_async_method, state_.GetStub(), &state_.GetClientContext(), &state_.GetQueue());
@@ -427,7 +427,7 @@ bool BidirectionalStream<Request, Response>::Write(const Request& request) {
 
     {
         const auto lock = state_.TakeMutexIfBidirectional();
-        RunMiddlewarePipeline(state_, SendMessageHooks(request));
+        RunMiddlewarePipeline(state_, MiddlewareHooks::SendMessageHooks(request));
     }
 
     // Don't buffer writes, optimize for ping-pong-style interaction
@@ -445,7 +445,7 @@ void BidirectionalStream<Request, Response>::WriteAndCheck(const Request& reques
 
     {
         const auto lock = state_.TakeMutexIfBidirectional();
-        RunMiddlewarePipeline(state_, SendMessageHooks(request));
+        RunMiddlewarePipeline(state_, MiddlewareHooks::SendMessageHooks(request));
     }
 
     // Don't buffer writes, optimize for ping-pong-style interaction
diff --git a/grpc/include/userver/ugrpc/client/impl/unary_call.hpp b/grpc/include/userver/ugrpc/client/impl/unary_call.hpp
@@ -163,10 +163,15 @@ class UnaryCall final {
         return AttemptCompletionStatus::kOk;
     }
 
-    void RunStartCallHooks() { impl::RunMiddlewarePipeline(state_, StartCallHooks(ToBaseMessage(&request_))); }
+    void RunStartCallHooks() {
+        impl::RunMiddlewarePipeline(state_, MiddlewareHooks::StartCallHooks(ToBaseMessage(&request_)));
+    }
 
     void RunFinishHooks(const grpc::Status& status) {
-        impl::RunMiddlewarePipeline(state_, FinishHooks(status, ToBaseMessage(&response_)));
+        impl::RunMiddlewarePipeline(
+            state_,
+            MiddlewareHooks::FinishHooks(status, status.ok() ? ToBaseMessage(&response_) : nullptr)
+        );
     }
 
     void OnDone(const grpc::Status& status) {
diff --git a/grpc/include/userver/ugrpc/client/stream_read_future.hpp b/grpc/include/userver/ugrpc/client/stream_read_future.hpp
@@ -107,7 +107,7 @@ bool StreamReadFuture<RawStream>::Get() {
         impl::Finish(*stream_, *state, /*final_response=*/nullptr, /*throw_on_error=*/true);
     } else {
         if (recv_message_) {
-            RunMiddlewarePipeline(*state, impl::RecvMessageHooks(*recv_message_));
+            RunMiddlewarePipeline(*state, impl::MiddlewareHooks::RecvMessageHooks(*recv_message_));
         }
     }
     return result == ugrpc::impl::AsyncMethodInvocation::WaitStatus::kOk;
diff --git a/grpc/src/ugrpc/client/impl/async_stream_methods.cpp b/grpc/src/ugrpc/client/impl/async_stream_methods.cpp
@@ -71,7 +71,7 @@ void ProcessFinish(CallState& state, const google::protobuf::Message* final_resp
 
     HandleCallStatistics(state, status);
 
-    RunMiddlewarePipeline(state, FinishHooks(status, final_response));
+    RunMiddlewarePipeline(state, MiddlewareHooks::FinishHooks(status, status.ok() ? final_response : nullptr));
 
     SetStatusAndResetSpan(state, status);
 }
diff --git a/grpc/src/ugrpc/client/impl/middleware_hooks.cpp b/grpc/src/ugrpc/client/impl/middleware_hooks.cpp
@@ -1,70 +1,61 @@
 #include <userver/ugrpc/client/impl/middleware_hooks.hpp>
 
+#include <userver/utils/assert.hpp>
+#include <userver/utils/overloaded.hpp>
+
 #include <userver/ugrpc/client/middlewares/base.hpp>
 
 USERVER_NAMESPACE_BEGIN
 
 namespace ugrpc::client::impl {
 
-void MiddlewareHooks::SetStartCall() noexcept { start_call_ = true; }
-
-void MiddlewareHooks::SetSendMessage(const google::protobuf::Message& send_message) noexcept {
-    send_message_ = &send_message;
+MiddlewareHooks MiddlewareHooks::StartCallHooks(const google::protobuf::Message* request) noexcept {
+    return MiddlewareHooks{Inbound{true, request}};
 }
 
-void MiddlewareHooks::SetRecvMessage(const google::protobuf::Message& recv_message) noexcept {
-    recv_message_ = &recv_message;
+MiddlewareHooks MiddlewareHooks::SendMessageHooks(const google::protobuf::Message& send_message) noexcept {
+    return MiddlewareHooks{Inbound{false, &send_message}};
 }
 
-void MiddlewareHooks::SetStatus(const grpc::Status& status) noexcept { status_ = &status; }
-
-void MiddlewareHooks::Run(const MiddlewareBase& middleware, MiddlewareCallContext& context) const {
-    if (start_call_) {
-        middleware.PreStartCall(context);
-    }
-
-    if (send_message_) {
-        middleware.PreSendMessage(context, *send_message_);
-    }
-
-    if (recv_message_) {
-        middleware.PostRecvMessage(context, *recv_message_);
-    }
-
-    if (status_) {
-        middleware.PostFinish(context, *status_);
-    }
+MiddlewareHooks MiddlewareHooks::FinishHooks(const grpc::Status& status, const google::protobuf::Message* response)
+    noexcept {
+    UASSERT(status.ok() || nullptr == response);
+    return MiddlewareHooks{Outbound{&status, response}};
 }
 
-MiddlewareHooks StartCallHooks(const google::protobuf::Message* request) noexcept {
-    MiddlewareHooks hooks;
-    hooks.SetStartCall();
-    if (request) {
-        hooks.SetSendMessage(*request);
-    }
-    return hooks;
+MiddlewareHooks MiddlewareHooks::RecvMessageHooks(const google::protobuf::Message& recv_message) noexcept {
+    return MiddlewareHooks{Outbound{nullptr, &recv_message}};
 }
 
-MiddlewareHooks SendMessageHooks(const google::protobuf::Message& send_message) noexcept {
-    MiddlewareHooks hooks;
-    hooks.SetSendMessage(send_message);
-    return hooks;
+void MiddlewareHooks::Run(const MiddlewareBase& middleware, MiddlewareCallContext& context) const {
+    std::visit(
+        utils::Overloaded{
+            [&middleware, &context](Inbound params) {
+                if (params.start_call) {
+                    middleware.PreStartCall(context);
+                }
+                if (params.send_message) {
+                    middleware.PreSendMessage(context, *params.send_message);
+                }
+            },
+            [&middleware, &context](Outbound params) {
+                if (params.recv_message) {
+                    middleware.PostRecvMessage(context, *params.recv_message);
+                }
+                if (params.status) {
+                    middleware.PostFinish(context, *params.status);
+                }
+            },
+        },
+        params_
+    );
 }
 
-MiddlewareHooks RecvMessageHooks(const google::protobuf::Message& recv_message) noexcept {
-    MiddlewareHooks hooks;
-    hooks.SetRecvMessage(recv_message);
-    return hooks;
-}
+bool MiddlewareHooks::Reverse() const noexcept { return std::holds_alternative<Outbound>(params_); }
 
-MiddlewareHooks FinishHooks(const grpc::Status& status, const google::protobuf::Message* response) noexcept {
-    MiddlewareHooks hooks;
-    if (status.ok() && response) {
-        hooks.SetRecvMessage(*response);
-    }
-    hooks.SetStatus(status);
-    return hooks;
-}
+MiddlewareHooks::MiddlewareHooks(Params&& params) noexcept
+        : params_{std::move(params)}
+{}
 
 }  // namespace ugrpc::client::impl
 
diff --git a/grpc/src/ugrpc/client/impl/middleware_pipeline.cpp b/grpc/src/ugrpc/client/impl/middleware_pipeline.cpp
@@ -1,5 +1,7 @@
 #include <userver/ugrpc/client/impl/middleware_pipeline.hpp>
 
+#include <boost/range/adaptor/reversed.hpp>
+
 #include <userver/logging/log.hpp>
 
 USERVER_NAMESPACE_BEGIN
@@ -8,8 +10,14 @@ namespace ugrpc::client::impl {
 
 void MiddlewarePipeline::Run(const MiddlewareHooks& hooks, MiddlewareCallContext& context) const {
     try {
-        for (const auto& m : middlewares_) {
-            hooks.Run(*m, context);
+        if (!hooks.Reverse()) {
+            for (const auto& m : middlewares_) {
+                hooks.Run(*m, context);
+            }
+        } else {
+            for (const auto& m : boost::adaptors::reverse(middlewares_)) {
+                hooks.Run(*m, context);
+            }
         }
     } catch (const std::exception& ex) {
         LOG_WARNING() << "Run middlewares failed: " << ex;

Original file line number	Diff line number	Diff line change
`@@ -257,7 +257,7 @@ InputStream<Response>::InputStream(`
`257`	`257`	`: state_{std::move(params), CallKind::kInputStream},`
`258`	`258`	`context_{utils::impl::InternalTag{}, state_}`
`259`	`259`	`{`
`260`		`- RunMiddlewarePipeline(state_, StartCallHooks(ToBaseMessage(&request)));`
	`260`	`+ RunMiddlewarePipeline(state_, MiddlewareHooks::StartCallHooks(ToBaseMessage(&request)));`
`261`	`261`
`262`	`262`	`// NOLINTNEXTLINE(cppcoreguidelines-prefer-member-initializer)`
`263`	`263`	`stream_ = impl::PrepareCall(`
`@@ -286,7 +286,7 @@ bool InputStream<Response>::Read(Response& response) {`
`286`	`286`	`}`
`287`	`287`
`288`	`288`	`if (impl::Read(*stream_, response, state_)) {`
`289`		`- RunMiddlewarePipeline(state_, RecvMessageHooks(response));`
	`289`	`+ RunMiddlewarePipeline(state_, MiddlewareHooks::RecvMessageHooks(response));`
`290`	`290`	`return true;`
`291`	`291`	`} else {`
`292`	`292`	`// Finish can only be called once all the data is read, otherwise the`
`@@ -305,7 +305,7 @@ OutputStream<Request, Response>::OutputStream(`
`305`	`305`	`: state_{std::move(params), CallKind::kOutputStream},`
`306`	`306`	`context_{utils::impl::InternalTag{}, state_}`
`307`	`307`	`{`
`308`		`- RunMiddlewarePipeline(state_, StartCallHooks());`
	`308`	`+ RunMiddlewarePipeline(state_, MiddlewareHooks::StartCallHooks());`
`309`	`309`
`310`	`310`	`// 'response_' will be filled upon successful 'Finish' async call`
`311`	`311`	`// NOLINTNEXTLINE(cppcoreguidelines-prefer-member-initializer)`
`@@ -332,7 +332,7 @@ bool OutputStream<Request, Response>::Write(const Request& request) {`
`332`	`332`	`return false;`
`333`	`333`	`}`
`334`	`334`
`335`		`- RunMiddlewarePipeline(state_, SendMessageHooks(request));`
	`335`	`+ RunMiddlewarePipeline(state_, MiddlewareHooks::SendMessageHooks(request));`
`336`	`336`
`337`	`337`	`// Don't buffer writes, otherwise in an event subscription scenario, events`
`338`	`338`	`// may never actually be delivered`
`@@ -348,7 +348,7 @@ void OutputStream<Request, Response>::WriteAndCheck(const Request& request) {`
`348`	`348`	`throw RpcError(state_.GetCallName(), "'WriteAndCheck' called on a finished or closed stream");`
`349`	`349`	`}`
`350`	`350`
`351`		`- RunMiddlewarePipeline(state_, SendMessageHooks(request));`
	`351`	`+ RunMiddlewarePipeline(state_, MiddlewareHooks::SendMessageHooks(request));`
`352`	`352`
`353`	`353`	`// Don't buffer writes, otherwise in an event subscription scenario, events`
`354`	`354`	`// may never actually be delivered`
`@@ -381,7 +381,7 @@ BidirectionalStream<Request, Response>::BidirectionalStream(`
`381`	`381`	`: state_{std::move(params), CallKind::kBidirectionalStream},`
`382`	`382`	`context_{utils::impl::InternalTag{}, state_}`
`383`	`383`	`{`
`384`		`- RunMiddlewarePipeline(state_, StartCallHooks());`
	`384`	`+ RunMiddlewarePipeline(state_, MiddlewareHooks::StartCallHooks());`
`385`	`385`
`386`	`386`	`// NOLINTNEXTLINE(cppcoreguidelines-prefer-member-initializer)`
`387`	`387`	`stream_ = impl::PrepareCall(prepare_async_method, state_.GetStub(), &state_.GetClientContext(), &state_.GetQueue());`
`@@ -427,7 +427,7 @@ bool BidirectionalStream<Request, Response>::Write(const Request& request) {`
`427`	`427`
`428`	`428`	`{`
`429`	`429`	`const auto lock = state_.TakeMutexIfBidirectional();`
`430`		`- RunMiddlewarePipeline(state_, SendMessageHooks(request));`
	`430`	`+ RunMiddlewarePipeline(state_, MiddlewareHooks::SendMessageHooks(request));`
`431`	`431`	`}`
`432`	`432`
`433`	`433`	`// Don't buffer writes, optimize for ping-pong-style interaction`
`@@ -445,7 +445,7 @@ void BidirectionalStream<Request, Response>::WriteAndCheck(const Request& reques`
`445`	`445`
`446`	`446`	`{`
`447`	`447`	`const auto lock = state_.TakeMutexIfBidirectional();`
`448`		`- RunMiddlewarePipeline(state_, SendMessageHooks(request));`
	`448`	`+ RunMiddlewarePipeline(state_, MiddlewareHooks::SendMessageHooks(request));`
`449`	`449`	`}`
`450`	`450`
`451`	`451`	`// Don't buffer writes, optimize for ping-pong-style interaction`
Original file line number	Diff line number	Diff line change
`@@ -107,7 +107,7 @@ bool StreamReadFuture<RawStream>::Get() {`
`107`	`107`	`impl::Finish(stream_, state, /final_response=/nullptr, /throw_on_error=/true);`
`108`	`108`	`} else {`
`109`	`109`	`if (recv_message_) {`
`110`		`- RunMiddlewarePipeline(state, impl::RecvMessageHooks(recv_message_));`
	`110`	`+ RunMiddlewarePipeline(state, impl::MiddlewareHooks::RecvMessageHooks(recv_message_));`
`111`	`111`	`}`
`112`	`112`	`}`
`113`	`113`	`return result == ugrpc::impl::AsyncMethodInvocation::WaitStatus::kOk;`
Original file line number	Diff line number	Diff line change
`@@ -71,7 +71,7 @@ void ProcessFinish(CallState& state, const google::protobuf::Message* final_resp`
`71`	`71`
`72`	`72`	`HandleCallStatistics(state, status);`
`73`	`73`
`74`		`- RunMiddlewarePipeline(state, FinishHooks(status, final_response));`
	`74`	`+ RunMiddlewarePipeline(state, MiddlewareHooks::FinishHooks(status, status.ok() ? final_response : nullptr));`
`75`	`75`
`76`	`76`	`SetStatusAndResetSpan(state, status);`
`77`	`77`	`}`