feat core: read abs tp from the new header field

commit_hash:46ecc8318a38e8c919fc4d43b58fcb2d1bc7b5a4

Author: oigolovanova

.mapping.json

@@ -652,6 +652,8 @@
   "core/dynamic_configs/USERVER_BAGGAGE_ENABLED.yaml":"taxi/uservices/userver/core/dynamic_configs/USERVER_BAGGAGE_ENABLED.yaml",
   "core/dynamic_configs/USERVER_CACHES.yaml":"taxi/uservices/userver/core/dynamic_configs/USERVER_CACHES.yaml",
   "core/dynamic_configs/USERVER_CANCEL_HANDLE_REQUEST_BY_DEADLINE.yaml":"taxi/uservices/userver/core/dynamic_configs/USERVER_CANCEL_HANDLE_REQUEST_BY_DEADLINE.yaml",
+  "core/dynamic_configs/USERVER_DEADLINE_PROPAGATION_ABSOLUTE_TIMESTAMP_ENABLED.yaml":"taxi/uservices/userver/core/dynamic_configs/USERVER_DEADLINE_PROPAGATION_ABSOLUTE_TIMESTAMP_ENABLED.yaml",
+  "core/dynamic_configs/USERVER_DEADLINE_PROPAGATION_CLOCK_SKEW_THRESHOLD_MS.yaml":"taxi/uservices/userver/core/dynamic_configs/USERVER_DEADLINE_PROPAGATION_CLOCK_SKEW_THRESHOLD_MS.yaml",
   "core/dynamic_configs/USERVER_DEADLINE_PROPAGATION_ENABLED.yaml":"taxi/uservices/userver/core/dynamic_configs/USERVER_DEADLINE_PROPAGATION_ENABLED.yaml",
   "core/dynamic_configs/USERVER_DUMPS.yaml":"taxi/uservices/userver/core/dynamic_configs/USERVER_DUMPS.yaml",
   "core/dynamic_configs/USERVER_FILES_CONTENT_TYPE_MAP.yaml":"taxi/uservices/userver/core/dynamic_configs/USERVER_FILES_CONTENT_TYPE_MAP.yaml",

core/dynamic_configs/USERVER_DEADLINE_PROPAGATION_ABSOLUTE_TIMESTAMP_ENABLED.yaml

@@ -0,0 +1,11 @@
+default: true
+description: |
+    When `false`, this service stops using the absolute instant from `X-Request-Deadline`
+    as the local request/task deadline, even if enabled in static config. Use for emergency
+    shutdown without restart.
+
+    The header may still be parsed and stored for forwarding to downstream services
+    (see `TaskInheritedData::original_deadline`). Duration-based propagation
+    (`X-YaTaxi-Client-TimeoutMs`) remains unaffected.
+schema:
+    type: boolean

core/dynamic_configs/USERVER_DEADLINE_PROPAGATION_CLOCK_SKEW_THRESHOLD_MS.yaml

@@ -0,0 +1,18 @@
+default: 60000
+description: |
+    Maximum allowed clock skew in milliseconds between the absolute deadline
+    timestamp and the expected value based on duration header.
+
+    When the detected skew exceeds this threshold, the absolute timestamp is
+    ignored and duration-based deadline propagation is used as a fallback.
+
+    Set to 0 to disable the check (always trust the absolute timestamp).
+
+    Recommended values:
+    - 60000 (1 minute) for well-synchronized environments (default)
+    - 300000 (5 minutes) for less reliable environments
+    - 0 only if you fully trust your clock synchronization
+schema:
+    type: integer
+    minimum: 0
+    x-taxi-cpp-type: std::chrono::milliseconds

core/functional_tests/basic_chaos/static_config.yaml

@@ -16,6 +16,7 @@ components_manager:
             path: /chaos/httpserver
             task_processor: main-task-processor
             method: GET,DELETE,POST
+            deadline_propagation_prefer_timestamp: true
 
         handler-chaos-httpserver-parse-body-args:
             path: /chaos/httpserver-parse-body-args

core/functional_tests/basic_chaos/tests-deadline/test_server.py

@@ -1,4 +1,5 @@
 import asyncio
+import datetime
 from typing import Any
 
 import pytest
@@ -11,6 +12,14 @@
 
 DP_TIMEOUT_MS = 'X-YaTaxi-Client-TimeoutMs'
 DP_DEADLINE_EXPIRED = 'X-YaTaxi-Deadline-Expired'
+DP_ABSOLUTE_DEADLINE = 'X-Request-Deadline'
+
+
+def _make_iso_deadline(offset_seconds: float) -> str:
+    tp = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
+        seconds=offset_seconds,
+    )
+    return tp.strftime('%Y-%m-%dT%H:%M:%S.') + f'{tp.microsecond:06d}Z'
 
 
 @pytest.fixture(name='call')
@@ -92,6 +101,95 @@ async def test_deadline_propagation_disabled_dynamically(call):
     assert response.status == 200
 
 
+async def test_absolute_deadline_used(call):
+    response = await call(
+        headers={
+            **HEADERS,
+            DP_TIMEOUT_MS: '1',
+            DP_ABSOLUTE_DEADLINE: _make_iso_deadline(5.0),
+        },
+    )
+    assert response.status == 200
+
+
+async def test_absolute_deadline_expired(call):
+    response = await call(
+        headers={
+            **HEADERS,
+            DP_ABSOLUTE_DEADLINE: _make_iso_deadline(-120.0),
+        },
+    )
+    _check_deadline_propagation_response(response)
+
+
+async def test_absolute_deadline_expired_with_sleep(call):
+    response = await call(
+        htype='sleep',
+        headers={
+            **HEADERS,
+            DP_ABSOLUTE_DEADLINE: _make_iso_deadline(-120.0),
+        },
+    )
+    _check_deadline_propagation_response(response)
+
+
+@pytest.mark.config(USERVER_DEADLINE_PROPAGATION_ABSOLUTE_TIMESTAMP_ENABLED=False)
+async def test_absolute_deadline_disabled_dynamically(call):
+    response = await call(
+        headers={
+            **HEADERS,
+            DP_TIMEOUT_MS: '5000',
+            DP_ABSOLUTE_DEADLINE: _make_iso_deadline(-120.0),
+        },
+    )
+    assert response.status == 200
+
+
+async def test_absolute_deadline_clock_skew_fallback(call):
+    response = await call(
+        headers={
+            **HEADERS,
+            DP_TIMEOUT_MS: '5000',
+            DP_ABSOLUTE_DEADLINE: _make_iso_deadline(5.0 + 120.0),
+        },
+    )
+    assert response.status == 200
+
+
+async def test_absolute_deadline_clock_skew_fallback_when_negative_skew(call):
+    response = await call(
+        headers={
+            **HEADERS,
+            DP_TIMEOUT_MS: '5000',
+            DP_ABSOLUTE_DEADLINE: _make_iso_deadline(-120.0),
+        },
+    )
+    assert response.status == 200
+
+
+@pytest.mark.config(USERVER_DEADLINE_PROPAGATION_CLOCK_SKEW_THRESHOLD_MS=0)
+async def test_absolute_deadline_threshold_zero_disables_skew_check(call):
+    response = await call(
+        headers={
+            **HEADERS,
+            DP_TIMEOUT_MS: '5000',
+            DP_ABSOLUTE_DEADLINE: _make_iso_deadline(-120.0),
+        },
+    )
+    _check_deadline_propagation_response(response)
+
+
+async def test_absolute_deadline_invalid_format(call):
+    response = await call(
+        headers={
+            **HEADERS,
+            DP_TIMEOUT_MS: '5000',
+            DP_ABSOLUTE_DEADLINE: 'not-a-timestamp',
+        },
+    )
+    assert response.status == 200
+
+
 async def test_cancellable(service_client, testpoint):
     @testpoint('testpoint_cancel')
     def cancel_testpoint(data):

core/library.yaml

@@ -17,6 +17,8 @@ configs:
       - USERVER_CACHES
       - USERVER_CANCEL_HANDLE_REQUEST_BY_DEADLINE
       - USERVER_DEADLINE_PROPAGATION_ENABLED
+      - USERVER_DEADLINE_PROPAGATION_ABSOLUTE_TIMESTAMP_ENABLED
+      - USERVER_DEADLINE_PROPAGATION_CLOCK_SKEW_THRESHOLD_MS
       - USERVER_DUMPS
       - USERVER_FILES_CONTENT_TYPE_MAP
       - USERVER_HTTP_PROXY

core/src/server/middlewares/deadline_propagation.cpp

@@ -14,6 +14,10 @@
 #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
@@ -60,6 +64,53 @@ 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) {
@@ -153,28 +204,57 @@ void DeadlinePropagation::SetupInheritedDeadline(
         return;
     }
 
+    const auto original_deadline = ParseAbsoluteDeadline(request);
+    if (original_deadline) {
+        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 (!timeout) {
-        return;
+
+    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) {
+            return;
+        }
+        chosen_deadline = engine::Deadline::FromTimePoint(request.GetStartTime() + *timeout);
+        span_deadline_received_ms = timeout->count();
     }
 
+    inherited_data.deadline = *chosen_deadline;
     dp_scope.need_log_response = config_snapshot[::dynamic_config::USERVER_LOG_REQUEST];
 
-    auto* span_opt = tracing::Span::CurrentSpanUnchecked();
     if (span_opt) {
-        span_opt->AddNonInheritableTag("deadline_received_ms", timeout->count());
+        span_opt->AddNonInheritableTag("deadline_received_ms", span_deadline_received_ms);
     }
 
-    const auto deadline = engine::Deadline::FromTimePoint(request.GetStartTime() + *timeout);
-    inherited_data.deadline = deadline;
-
-    if (deadline.IsSurelyReachedApprox()) {
+    if (chosen_deadline->IsSurelyReachedApprox()) {
         HandleDeadlineExpired(request, dp_scope, "Immediate timeout (deadline propagation)");
         return;
     }
 
     if (config_snapshot[::dynamic_config::USERVER_CANCEL_HANDLE_REQUEST_BY_DEADLINE]) {
-        engine::current_task::SetDeadline(deadline);
+        engine::current_task::SetDeadline(*chosen_deadline);
     }
 }