fix grpc: fix channel/context Qos config race

commit_hash:0df875ca4d23d8c51dbd4554a48ad6e1e95f27e5

Author: kpavlov00

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

@@ -59,11 +59,9 @@ class ChannelArgumentsBuilder final {
     ChannelArgumentsBuilder& operator=(const ChannelArgumentsBuilder&) = delete;
 
     grpc::ChannelArguments Build(const ClientQos& client_qos) const;
-    grpc::ChannelArguments Build() const;
 
 private:
     const grpc::ChannelArguments& channel_args_;
-    const std::optional<std::string>& static_service_config_;
 
     ServiceConfigBuilder service_config_builder_;
 };

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

@@ -12,6 +12,7 @@
 #include <userver/testsuite/grpc_control.hpp>
 #include <userver/utils/fixed_array.hpp>
 
+#include <userver/ugrpc/client/client_qos.hpp>
 #include <userver/ugrpc/client/impl/channel_arguments_builder.hpp>
 #include <userver/ugrpc/client/impl/client_internals.hpp>
 #include <userver/ugrpc/client/impl/stub_any.hpp>
@@ -32,6 +33,8 @@ struct GenericClientTag final {
 class ClientData final {
 public:
     struct StubState {
+        ClientQos client_qos;
+
         StubPool stubs;
         // method_id -> stub_pool
         utils::FixedArray<StubPool> dedicated_stubs;
@@ -52,8 +55,10 @@ class ClientData final {
             return StubCast<Stub>(stub_);
         }
 
+        const ClientQos& GetClientQos() const { return state_->client_qos; }
+
     private:
-        rcu::ReadablePtr<StubState> state_;
+        const rcu::ReadablePtr<StubState> state_;
         StubAny& stub_;
     };
 
@@ -74,14 +79,14 @@ class ClientData final {
         if (internals_.qos) {
             SubscribeOnConfigUpdate<Service>(*internals_.qos);
         } else {
-            ConstructStubState<Service>(channel_arguments_builder_->Build());
+            ConstructStubState<Service>();
         }
     }
 
     template <typename Service>
     ClientData(ClientInternals&& internals, GenericClientTag, std::in_place_type_t<Service>)
         : internals_(std::move(internals)), stub_state_(std::make_unique<rcu::Variable<StubState>>()) {
-        ConstructStubState<Service>(BuildChannelArguments(internals_.channel_args, internals_.default_service_config));
+        ConstructStubState<Service>();
     }
 
     ~ClientData();
@@ -154,12 +159,16 @@ class ClientData final {
     void OnConfigUpdate(const dynamic_config::Snapshot& config) {
         UASSERT(internals_.qos);
         const auto& client_qos = config[*internals_.qos];
-        UASSERT(channel_arguments_builder_);
-        ConstructStubState<Service>(channel_arguments_builder_->Build(client_qos));
+        ConstructStubState<Service>(client_qos);
     }
 
     template <typename Service>
-    void ConstructStubState(const grpc::ChannelArguments& channel_args) {
+    void ConstructStubState(const ClientQos& client_qos = {}) {
+        const auto channel_args =
+            channel_arguments_builder_.has_value()
+                ? channel_arguments_builder_->Build(client_qos)
+                : BuildChannelArguments(internals_.channel_args, internals_.default_service_config);
+
         auto stubs = StubPool::Create<typename Service::Stub>(
             internals_.channel_count, internals_.channel_factory, channel_args
         );
@@ -171,7 +180,7 @@ class ClientData final {
                   )
                 : utils::FixedArray<StubPool>{};
 
-        stub_state_->Assign({std::move(stubs), std::move(dedicated_stubs)});
+        stub_state_->Assign({client_qos, std::move(stubs), std::move(dedicated_stubs)});
     }
 
     ClientInternals internals_;

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

@@ -7,6 +7,7 @@
 
 #include <grpcpp/completion_queue.h>
 
+#include <userver/ugrpc/client/client_qos.hpp>
 #include <userver/ugrpc/client/impl/client_data.hpp>
 #include <userver/ugrpc/client/middlewares/fwd.hpp>
 #include <userver/ugrpc/client/qos.hpp>

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

@@ -8,5 +8,4 @@
 #include <string_view>
 #include <utility>
 
-#include <userver/ugrpc/client/client_qos.hpp>
 #include <userver/ugrpc/impl/protobuf_collector.hpp>

grpc/src/ugrpc/client/impl/call_params.cpp

@@ -43,16 +43,16 @@ void SetDeadline(grpc::ClientContext& client_context, const Qos& qos, const test
 // 2. manual client_context manipulation by the user
 // 3. GRPC_CLIENT_QOS dynamic config
 void ApplyQosConfigs(
-    const ClientData& client_data,
     grpc::ClientContext& client_context,
     const Qos& user_qos,
-    std::string_view call_name
+    const Qos& dynamic_qos,
+    const testsuite::GrpcControl& testsuite_grpc
 ) {
     if (user_qos.timeout) {
         // Consider the explicit Qos parameter the highest-priority source.
         // TODO there is no way to override other sources by setting this timeout
         // to infinity (we treat it as "not set")
-        SetDeadline(client_context, user_qos, client_data.GetTestsuiteControl());
+        SetDeadline(client_context, user_qos, testsuite_grpc);
         return;
     }
 
@@ -62,12 +62,7 @@ void ApplyQosConfigs(
         return;
     }
 
-    if (const auto* const config_key = client_data.GetClientQos()) {
-        const auto config = client_data.GetConfigSnapshot();
-        if (const auto dynamic_qos = config[*config_key].GetOptional(call_name)) {
-            SetDeadline(client_context, *dynamic_qos, client_data.GetTestsuiteControl());
-        }
-    }
+    SetDeadline(client_context, dynamic_qos, testsuite_grpc);
 }
 
 }  // namespace
@@ -85,14 +80,16 @@ CallParams CreateCallParams(
         throw RpcCancelledError(call_name, "RPC construction");
     }
 
-    ApplyQosConfigs(client_data, *client_context, qos, call_name);
+    auto stub = client_data.NextStubFromMethodId(method_id);
+    const auto dynamic_qos = stub.GetClientQos().GetOptional(call_name).value_or(Qos{});
+    ApplyQosConfigs(*client_context, qos, dynamic_qos, client_data.GetTestsuiteControl());
 
     return CallParams{
         client_data.GetClientName(),  //
         client_data.NextQueue(),
         client_data.GetConfigSnapshot(),
         {ugrpc::impl::MaybeOwnedString::Ref{}, call_name},
-        client_data.NextStubFromMethodId(method_id),
+        std::move(stub),
         std::move(client_context),
         client_data.GetStatistics(method_id),
         client_data.GetMiddlewares(),
@@ -115,7 +112,8 @@ CallParams CreateGenericCallParams(
         throw RpcCancelledError(call_name, "RPC construction");
     }
 
-    ApplyQosConfigs(client_data, *client_context, qos, call_name);
+    UINVARIANT(!client_data.GetClientQos(), "Client QOS configs are unsupported for generic services");
+    ApplyQosConfigs(*client_context, qos, {}, client_data.GetTestsuiteControl());
 
     return CallParams{
         client_data.GetClientName(),  //

grpc/src/ugrpc/client/impl/channel_arguments_builder.cpp

@@ -231,9 +231,7 @@ ChannelArgumentsBuilder::ChannelArgumentsBuilder(
     const std::optional<std::string>& static_service_config,
     const ugrpc::impl::StaticServiceMetadata& metadata
 )
-    : channel_args_{channel_args},
-      static_service_config_{static_service_config},
-      service_config_builder_{metadata, static_service_config} {}
+    : channel_args_{channel_args}, service_config_builder_{metadata, static_service_config} {}
 
 grpc::ChannelArguments ChannelArgumentsBuilder::Build(const ClientQos& client_qos) const {
     const auto service_config = service_config_builder_.Build(client_qos);
@@ -243,10 +241,6 @@ grpc::ChannelArguments ChannelArgumentsBuilder::Build(const ClientQos& client_qo
     return BuildChannelArguments(channel_args_, formats::json::ToString(service_config));
 }
 
-grpc::ChannelArguments ChannelArgumentsBuilder::Build() const {
-    return BuildChannelArguments(channel_args_, static_service_config_);
-}
-
 grpc::ChannelArguments
 BuildChannelArguments(const grpc::ChannelArguments& channel_args, const std::optional<std::string>& service_config) {
     if (!service_config.has_value()) {