feat grpc: read abs tp from the new header field

commit_hash:b7a9cf9862d6aca03992007efa9f5f7e117ee01e

Author: oigolovanova

.mapping.json

@@ -1148,6 +1148,7 @@
   "core/include/userver/server/middlewares/cors.hpp":"taxi/uservices/userver/core/include/userver/server/middlewares/cors.hpp",
   "core/include/userver/server/middlewares/headers_propagator.hpp":"taxi/uservices/userver/core/include/userver/server/middlewares/headers_propagator.hpp",
   "core/include/userver/server/middlewares/http_middleware_base.hpp":"taxi/uservices/userver/core/include/userver/server/middlewares/http_middleware_base.hpp",
+  "core/include/userver/server/request/impl/absolute_deadline.hpp":"taxi/uservices/userver/core/include/userver/server/request/impl/absolute_deadline.hpp",
   "core/include/userver/server/request/json_data.hpp":"taxi/uservices/userver/core/include/userver/server/request/json_data.hpp",
   "core/include/userver/server/request/request_config.hpp":"taxi/uservices/userver/core/include/userver/server/request/request_config.hpp",
   "core/include/userver/server/request/request_context.hpp":"taxi/uservices/userver/core/include/userver/server/request/request_context.hpp",
@@ -2027,6 +2028,7 @@
   "core/src/server/net/listener_impl.hpp":"taxi/uservices/userver/core/src/server/net/listener_impl.hpp",
   "core/src/server/net/stats.hpp":"taxi/uservices/userver/core/src/server/net/stats.hpp",
   "core/src/server/pph_config.hpp":"taxi/uservices/userver/core/src/server/pph_config.hpp",
+  "core/src/server/request/impl/absolute_deadline.cpp":"taxi/uservices/userver/core/src/server/request/impl/absolute_deadline.cpp",
   "core/src/server/request/internal_request_context.cpp":"taxi/uservices/userver/core/src/server/request/internal_request_context.cpp",
   "core/src/server/request/internal_request_context.hpp":"taxi/uservices/userver/core/src/server/request/internal_request_context.hpp",
   "core/src/server/request/json_data.cpp":"taxi/uservices/userver/core/src/server/request/json_data.cpp",
@@ -2287,6 +2289,7 @@
   "grpc/functional_tests/basic_chaos/tests-grpcserver/test_server_smaller_parts.py":"taxi/uservices/userver/grpc/functional_tests/basic_chaos/tests-grpcserver/test_server_smaller_parts.py",
   "grpc/functional_tests/basic_chaos/tests-nonchaos/conftest.py":"taxi/uservices/userver/grpc/functional_tests/basic_chaos/tests-nonchaos/conftest.py",
   "grpc/functional_tests/basic_chaos/tests-nonchaos/test_basic.py":"taxi/uservices/userver/grpc/functional_tests/basic_chaos/tests-nonchaos/test_basic.py",
+  "grpc/functional_tests/basic_chaos/tests-nonchaos/test_deadline_absolute.py":"taxi/uservices/userver/grpc/functional_tests/basic_chaos/tests-nonchaos/test_deadline_absolute.py",
   "grpc/functional_tests/basic_chaos/tests-nonchaos/test_tracing_metadata.py":"taxi/uservices/userver/grpc/functional_tests/basic_chaos/tests-nonchaos/test_tracing_metadata.py",
   "grpc/functional_tests/basic_server/CMakeLists.txt":"taxi/uservices/userver/grpc/functional_tests/basic_server/CMakeLists.txt",
   "grpc/functional_tests/basic_server/grpc_service.cpp":"taxi/uservices/userver/grpc/functional_tests/basic_server/grpc_service.cpp",
@@ -2662,6 +2665,8 @@
   "grpc/src/ugrpc/server/middlewares/congestion_control/component.yaml":"taxi/uservices/userver/grpc/src/ugrpc/server/middlewares/congestion_control/component.yaml",
   "grpc/src/ugrpc/server/middlewares/congestion_control/middleware.cpp":"taxi/uservices/userver/grpc/src/ugrpc/server/middlewares/congestion_control/middleware.cpp",
   "grpc/src/ugrpc/server/middlewares/congestion_control/middleware.hpp":"taxi/uservices/userver/grpc/src/ugrpc/server/middlewares/congestion_control/middleware.hpp",
+  "grpc/src/ugrpc/server/middlewares/deadline_propagation/component.cpp":"taxi/uservices/userver/grpc/src/ugrpc/server/middlewares/deadline_propagation/component.cpp",
+  "grpc/src/ugrpc/server/middlewares/deadline_propagation/component.yaml":"taxi/uservices/userver/grpc/src/ugrpc/server/middlewares/deadline_propagation/component.yaml",
   "grpc/src/ugrpc/server/middlewares/deadline_propagation/middleware.cpp":"taxi/uservices/userver/grpc/src/ugrpc/server/middlewares/deadline_propagation/middleware.cpp",
   "grpc/src/ugrpc/server/middlewares/graceful_shutdown_headers/component.cpp":"taxi/uservices/userver/grpc/src/ugrpc/server/middlewares/graceful_shutdown_headers/component.cpp",
   "grpc/src/ugrpc/server/middlewares/graceful_shutdown_headers/middleware.cpp":"taxi/uservices/userver/grpc/src/ugrpc/server/middlewares/graceful_shutdown_headers/middleware.cpp",

core/include/userver/server/request/impl/absolute_deadline.hpp

@@ -0,0 +1,36 @@
+#pragma once
+
+/// @file userver/server/request/impl/absolute_deadline.hpp
+/// @brief Absolute deadline parsing and engine deadline construction for propagation.
+
+#include <chrono>
+#include <optional>
+#include <string>
+
+#include <userver/dynamic_config/fwd.hpp>
+#include <userver/engine/deadline.hpp>
+#include <userver/tracing/fwd.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace server::request {
+
+using TaskInheritedOriginalDeadline = std::chrono::time_point<std::chrono::system_clock, std::chrono::microseconds>;
+
+namespace impl {
+
+std::optional<TaskInheritedOriginalDeadline> ParseXRequestDeadlineString(const std::string& timestring);
+
+std::optional<engine::Deadline> TryMakeDeadlinePreferringAbsoluteTimestamp(
+    const std::optional<TaskInheritedOriginalDeadline>& absolute_original_deadline,
+    bool deadline_propagation_prefer_timestamp,
+    const std::optional<std::chrono::milliseconds>& client_deadline_duration,
+    const dynamic_config::Snapshot& config,
+    tracing::Span* span
+);
+
+}  // namespace impl
+
+}  // namespace server::request
+
+USERVER_NAMESPACE_END

core/include/userver/server/request/task_inherited_data.hpp

@@ -6,19 +6,17 @@
 #include <atomic>
 #include <chrono>
 #include <optional>
-#include <string>
+#include <string_view>
 
 #include <userver/engine/deadline.hpp>
 #include <userver/engine/task/inherited_variable.hpp>
+#include <userver/server/request/impl/absolute_deadline.hpp>
 
 USERVER_NAMESPACE_BEGIN
 
 /// Server request related types and functions
 namespace server::request {
 
-/// Microsecond-precision system-clock instant (e.g. from `X-Request-Deadline`).
-using TaskInheritedOriginalDeadline = std::chrono::time_point<std::chrono::system_clock, std::chrono::microseconds>;
-
 /// @brief Signals when an operation has detected deadline expiration.
 class DeadlineSignal final {
 public:

core/include/userver/tracing/span.hpp

@@ -277,6 +277,8 @@ class Span final {
 
     std::chrono::system_clock::time_point GetStartSystemTime() const;
 
+    std::chrono::steady_clock::time_point GetStartSteadyTime() const;
+
     /// @cond
     // For internal use only.
     void AddTags(const logging::LogExtra&, utils::impl::InternalTag);

core/src/server/middlewares/deadline_propagation.cpp

@@ -3,6 +3,7 @@
 #include <server/handlers/http_server_settings.hpp>
 #include <server/request/internal_request_context.hpp>
 
+#include <userver/engine/deadline.hpp>
 #include <userver/http/common_headers.hpp>
 #include <userver/server/handlers/exceptions.hpp>
 #include <userver/server/handlers/http_handler_base.hpp>
@@ -14,10 +15,6 @@
 #include <userver/utils/from_string.hpp>
 #include <userver/utils/overloaded.hpp>
 
-#include <userver/utils/datetime.hpp>
-
-#include <dynamic_config/variables/USERVER_DEADLINE_PROPAGATION_ABSOLUTE_TIMESTAMP_ENABLED.hpp>
-#include <dynamic_config/variables/USERVER_DEADLINE_PROPAGATION_CLOCK_SKEW_THRESHOLD_MS.hpp>
 #include <dynamic_config/variables/USERVER_DEADLINE_PROPAGATION_ENABLED.hpp>
 
 USERVER_NAMESPACE_BEGIN
@@ -64,53 +61,6 @@ std::optional<std::chrono::milliseconds> ParseTimeout(const http::HttpRequest& r
     return timeout;
 }
 
-std::optional<request::TaskInheritedOriginalDeadline> ParseAbsoluteDeadline(const http::HttpRequest& request) {
-    const auto& header_str = request.GetHeader(USERVER_NAMESPACE::http::headers::kXRequestDeadline);
-    if (header_str.empty()) {
-        return std::nullopt;
-    }
-
-    LOG_DEBUG() << "Got X-Request-Deadline: " << header_str;
-    try {
-        const auto timestamp = utils::datetime::UtcStringtime(header_str, utils::datetime::kTaximeterFormat);
-        return std::chrono::time_point_cast<std::chrono::microseconds>(timestamp);
-    } catch (const std::exception& exception) {
-        LOG_LIMITED_WARNING() << "Can't parse X-Request-Deadline from '" << header_str << "': " << exception.what();
-        return std::nullopt;
-    }
-}
-
-/// Returns true if clock skew is within the threshold (absolute deadline is trustworthy).
-bool IsClockSkewAcceptable(
-    const request::TaskInheritedOriginalDeadline& absolute_tp,
-    std::chrono::milliseconds duration_timeout,
-    std::chrono::milliseconds threshold,
-    tracing::Span* span
-) {
-    if (threshold == std::chrono::milliseconds{0}) {
-        return true;  // Check disabled
-    }
-
-    const auto expected_timestamp =
-        std::chrono::time_point_cast<std::chrono::microseconds>(utils::datetime::Now()) + duration_timeout;
-
-    const auto skew = std::chrono::abs(absolute_tp - expected_timestamp);
-    const auto skew_ms = std::chrono::duration_cast<std::chrono::milliseconds>(skew);
-
-    if (span) {
-        span->AddNonInheritableTag("dp_clock_skew_ms", skew_ms.count());
-    }
-
-    if (skew_ms > threshold) {
-        LOG_LIMITED_WARNING()
-            << "Clock skew detected: " << skew_ms << ", threshold: " << threshold
-            << ". Falling back to duration-based deadline propagation";
-        return false;
-    }
-
-    return true;
-}
-
 void SetFormattedErrorResponse(http::HttpResponse& http_response, handlers::FormattedErrorData&& formatted_error_data) {
     http_response.SetData(std::move(formatted_error_data.external_body));
     if (formatted_error_data.content_type) {
@@ -204,48 +154,46 @@ void DeadlinePropagation::SetupInheritedDeadline(
         return;
     }
 
-    const auto original_deadline = ParseAbsoluteDeadline(request);
-    if (original_deadline) {
+    const auto& header_str = request.GetHeader(USERVER_NAMESPACE::http::headers::kXRequestDeadline);
+    if (!header_str.empty()) {
+        LOG_DEBUG() << "Got X-Request-Deadline: " << header_str;
+    }
+    const auto original_deadline = request::impl::ParseXRequestDeadlineString(header_str);
+    if (original_deadline.has_value()) {
         inherited_data.original_deadline = original_deadline;
     }
 
-    std::optional<engine::Deadline> chosen_deadline;
-    std::int64_t span_deadline_received_ms = 0;
     auto* span_opt = tracing::Span::CurrentSpanUnchecked();
     const auto timeout = ParseTimeout(request);
 
-    if (original_deadline.has_value() && deadline_propagation_prefer_timestamp_ &&
-        config_snapshot[::dynamic_config::USERVER_DEADLINE_PROPAGATION_ABSOLUTE_TIMESTAMP_ENABLED])
-    {
-        bool use_absolute = true;
-        if (timeout.has_value()) {
-            const auto&
-                threshold = config_snapshot[::dynamic_config::USERVER_DEADLINE_PROPAGATION_CLOCK_SKEW_THRESHOLD_MS];
-            use_absolute = IsClockSkewAcceptable(*original_deadline, *timeout, threshold, span_opt);
-        }
-
-        if (use_absolute) {
-            chosen_deadline = engine::Deadline::FromTimePoint(*original_deadline);
-
-            const auto now_sys_micro = std::chrono::time_point_cast<std::chrono::microseconds>(utils::datetime::Now());
-            span_deadline_received_ms =
-                std::chrono::duration_cast<std::chrono::milliseconds>(*original_deadline - now_sys_micro).count();
-        }
-    }
-
-    if (!chosen_deadline) {
-        if (!timeout) {
+    std::optional<engine::Deadline> chosen_deadline = request::impl::TryMakeDeadlinePreferringAbsoluteTimestamp(
+        original_deadline,
+        deadline_propagation_prefer_timestamp_,
+        timeout,
+        config_snapshot,
+        span_opt
+    );
+
+    std::chrono::milliseconds span_deadline_received{0};
+    if (!chosen_deadline.has_value()) {
+        if (!timeout.has_value()) {
             return;
         }
         chosen_deadline = engine::Deadline::FromTimePoint(request.GetStartTime() + *timeout);
-        span_deadline_received_ms = timeout->count();
+        span_deadline_received = *timeout;
+    } else {
+        span_deadline_received =
+            *chosen_deadline == engine::Deadline::Passed()
+                ? std::chrono::milliseconds{0}
+                : std::chrono::duration_cast<
+                      std::chrono::milliseconds>(chosen_deadline->GetTimePoint() - request.GetStartTime());
     }
 
     inherited_data.deadline = *chosen_deadline;
     dp_scope.need_log_response = config_snapshot[::dynamic_config::USERVER_LOG_REQUEST];
 
     if (span_opt) {
-        span_opt->AddNonInheritableTag("deadline_received_ms", span_deadline_received_ms);
+        span_opt->AddNonInheritableTag("deadline_received_ms", span_deadline_received.count());
     }
 
     if (chosen_deadline->IsSurelyReachedApprox()) {

core/src/server/request/impl/absolute_deadline.cpp

@@ -0,0 +1,96 @@
+#include <userver/server/request/impl/absolute_deadline.hpp>
+
+#include <string>
+
+#include <userver/dynamic_config/snapshot.hpp>
+#include <userver/logging/log.hpp>
+#include <userver/tracing/span.hpp>
+#include <userver/utils/datetime.hpp>
+
+#include <dynamic_config/variables/USERVER_DEADLINE_PROPAGATION_ABSOLUTE_TIMESTAMP_ENABLED.hpp>
+#include <dynamic_config/variables/USERVER_DEADLINE_PROPAGATION_CLOCK_SKEW_THRESHOLD_MS.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace server::request::impl {
+
+namespace {
+
+bool IsClockSkewAcceptable(
+    const TaskInheritedOriginalDeadline& absolute_tp,
+    std::chrono::milliseconds deadline_duration,
+    std::chrono::milliseconds threshold,
+    tracing::Span* span
+) {
+    if (threshold == std::chrono::milliseconds{0}) {
+        return true;
+    }
+
+    const auto request_start_system_time = span ? span->GetStartSystemTime() : utils::datetime::Now();
+    const auto expected_timestamp =
+        std::chrono::time_point_cast<std::chrono::microseconds>(request_start_system_time) + deadline_duration;
+
+    const auto skew = std::chrono::abs(absolute_tp - expected_timestamp);
+    const auto skew_ms = std::chrono::duration_cast<std::chrono::milliseconds>(skew);
+
+    if (span != nullptr) {
+        span->AddNonInheritableTag("dp_clock_skew_ms", skew_ms.count());
+    }
+
+    if (skew_ms > threshold) {
+        LOG_LIMITED_WARNING()
+            << "Clock skew detected: " << skew_ms << ", threshold: " << threshold
+            << ". Falling back to duration-based deadline propagation";
+        return false;
+    }
+
+    return true;
+}
+
+}  // namespace
+
+std::optional<TaskInheritedOriginalDeadline> ParseXRequestDeadlineString(const std::string& timestring) {
+    if (timestring.empty()) {
+        return std::nullopt;
+    }
+    try {
+        const auto timestamp = utils::datetime::UtcStringtime(timestring, utils::datetime::kTaximeterFormat);
+        return std::chrono::time_point_cast<std::chrono::microseconds>(timestamp);
+    } catch (const std::exception& exception) {
+        LOG_LIMITED_WARNING() << "Can't parse X-Request-Deadline: " << exception.what();
+        return std::nullopt;
+    }
+}
+
+std::optional<engine::Deadline> TryMakeDeadlinePreferringAbsoluteTimestamp(
+    const std::optional<TaskInheritedOriginalDeadline>& absolute_original_deadline,
+    const bool deadline_propagation_prefer_timestamp,
+    const std::optional<std::chrono::milliseconds>& client_deadline_duration,
+    const dynamic_config::Snapshot& config,
+    tracing::Span* span
+) {
+    if (!absolute_original_deadline.has_value() || !deadline_propagation_prefer_timestamp ||
+        !config[::dynamic_config::USERVER_DEADLINE_PROPAGATION_ABSOLUTE_TIMESTAMP_ENABLED])
+    {
+        return std::nullopt;
+    }
+
+    bool use_absolute = true;
+    if (client_deadline_duration.has_value()) {
+        use_absolute = IsClockSkewAcceptable(
+            *absolute_original_deadline,
+            *client_deadline_duration,
+            config[::dynamic_config::USERVER_DEADLINE_PROPAGATION_CLOCK_SKEW_THRESHOLD_MS],
+            span
+        );
+    }
+
+    if (use_absolute) {
+        return engine::Deadline::FromTimePoint(*absolute_original_deadline);
+    }
+    return std::nullopt;
+}
+
+}  // namespace server::request::impl
+
+USERVER_NAMESPACE_END

core/src/tracing/span.cpp

@@ -434,6 +434,8 @@ void Span::AttachToCoroStack() { pimpl_->AttachToCoroStack(); }
 
 std::chrono::system_clock::time_point Span::GetStartSystemTime() const { return pimpl_->start_system_time_; }
 
+std::chrono::steady_clock::time_point Span::GetStartSteadyTime() const { return pimpl_->start_steady_time_; }
+
 std::string_view Span::GetTraceId() const { return pimpl_->GetTraceId(); }
 
 std::string_view Span::GetSpanId() const { return pimpl_->GetSpanId(); }

grpc/CMakeLists.txt

@@ -47,6 +47,7 @@ list(
     src/ugrpc/client/simple_client_component.yaml
     src/ugrpc/server/middlewares/access_log/component.yaml
     src/ugrpc/server/middlewares/congestion_control/component.yaml
+    src/ugrpc/server/middlewares/deadline_propagation/component.yaml
     src/ugrpc/server/middlewares/headers_propagator/component.yaml
     src/ugrpc/server/middlewares/log/component.yaml
     src/ugrpc/server/server_component.yaml

grpc/functional_tests/basic_chaos/main.cpp

@@ -4,6 +4,7 @@
 #include <userver/clients/http/component_list.hpp>
 #include <userver/components/minimal_server_component_list.hpp>
 #include <userver/congestion_control/component.hpp>
+#include <userver/dynamic_config/updater/component_list.hpp>
 #include <userver/server/handlers/tests_control.hpp>
 #include <userver/testsuite/testsuite_support.hpp>
 #include <userver/ugrpc/client/client_factory_component.hpp>
@@ -18,6 +19,7 @@
 int main(int argc, char* argv[]) {
     const auto component_list =
         components::MinimalServerComponentList()
+            .AppendComponentList(USERVER_NAMESPACE::dynamic_config::updater::ComponentList())
             .Append<components::TestsuiteSupport>()
             .AppendComponentList(ugrpc::server::DefaultComponentList())
             .Append<ugrpc::server::middlewares::access_log::Component>()