From 0ca21492c99752bb2159b2ccee99270da821b2fb Mon Sep 17 00:00:00 2001 From: Alexander Polyakov Date: Mon, 10 Aug 2026 17:21:59 +0300 Subject: [PATCH 1/9] add confdata-proxy-actor-name argument for confdata component --- .../components/confdata/bindings/bindings.cpp | 7 +++-- .../components/confdata/confdata.cmake | 1 + .../confdata/state/component-state.cpp | 28 ++++++++++++++++++ .../confdata/state/component-state.h | 29 ++++++++++++++++++- 4 files changed, 62 insertions(+), 3 deletions(-) create mode 100644 runtime-light/components/confdata/state/component-state.cpp diff --git a/runtime-light/components/confdata/bindings/bindings.cpp b/runtime-light/components/confdata/bindings/bindings.cpp index 5d3ae4366a..30569fcd5a 100644 --- a/runtime-light/components/confdata/bindings/bindings.cpp +++ b/runtime-light/components/confdata/bindings/bindings.cpp @@ -7,6 +7,7 @@ #include "runtime-common/core/runtime-core.h" #include "runtime-light/allocator/allocator-state.h" +#include "runtime-light/components/confdata/state/component-state.h" #include "runtime-light/components/confdata/state/instance-state.h" #include "runtime-light/coroutine/coroutine-state.h" #include "runtime-light/coroutine/io-scheduler.h" @@ -39,8 +40,10 @@ auto contextual_tags::try_get() noexcept -> std::optional const AllocatorState& { - if (k2::instance_state() != nullptr) [[likely]] { - return InstanceState::get().instance_allocator_state; + if (const auto* instance_state_ptr{k2::instance_state()}; instance_state_ptr != nullptr) [[likely]] { + return instance_state_ptr->instance_allocator_state; + } else if (const auto* component_state_ptr{k2::component_state()}; component_state_ptr != nullptr) { + return component_state_ptr->m_allocator_state; } kphp::log::error("can't find allocator state"); } diff --git a/runtime-light/components/confdata/confdata.cmake b/runtime-light/components/confdata/confdata.cmake index a6f2478d31..296c2467c2 100644 --- a/runtime-light/components/confdata/confdata.cmake +++ b/runtime-light/components/confdata/confdata.cmake @@ -3,6 +3,7 @@ set(K2_CONFDATA_COMPONENT_SRC ${RUNTIME_LIGHT_DIR}/components/confdata/confdata-component.cpp ${RUNTIME_LIGHT_DIR}/components/confdata/bindings/bindings.cpp + ${RUNTIME_LIGHT_DIR}/components/confdata/state/component-state.cpp ${RUNTIME_LIGHT_DIR}/components/confdata/state/instance-state.cpp) set(K2_CONFDATA_ALLOCATOR_SRC diff --git a/runtime-light/components/confdata/state/component-state.cpp b/runtime-light/components/confdata/state/component-state.cpp new file mode 100644 index 0000000000..335a56a1eb --- /dev/null +++ b/runtime-light/components/confdata/state/component-state.cpp @@ -0,0 +1,28 @@ +// Compiler for PHP (aka KPHP) +// Copyright (c) 2026 LLC «V Kontakte» +// Distributed under the GPL v3 License, see LICENSE.notice.txt + +#include "runtime-light/components/confdata/state/component-state.h" + +#include + +#include "runtime-light/k2-platform/k2-api.h" +#include "runtime-light/stdlib/diagnostics/logs.h" + +auto ComponentState::parse_confdata_proxy_actor_name_arg(std::string_view value_view) noexcept -> void { + m_confdata_proxy_actor_name = value_view; +} + +auto ComponentState::parse_args() noexcept -> void { + for (auto i = 0; i < m_argc; ++i) { + const auto [arg_key, arg_value]{k2::arg_fetch(i)}; + const std::string_view key_view{arg_key.get(), std::strlen(arg_key.get())}; + const std::string_view value_view{arg_value.get(), std::strlen(arg_value.get())}; + + if (key_view == CONFDATA_PROXY_ACTOR_NAME_ARG) { + parse_confdata_proxy_actor_name_arg(value_view); + } else { + kphp::log::error("unexpected argument: {}", key_view); + } + } +} diff --git a/runtime-light/components/confdata/state/component-state.h b/runtime-light/components/confdata/state/component-state.h index c5e3059aa6..9e39724264 100644 --- a/runtime-light/components/confdata/state/component-state.h +++ b/runtime-light/components/confdata/state/component-state.h @@ -5,16 +5,43 @@ #pragma once #include +#include #include "common/mixin/not_copyable.h" +#include "runtime-common/core/allocator/script-allocator.h" +#include "runtime-common/core/std/containers.h" +#include "runtime-light/allocator/allocator-state.h" #include "runtime-light/k2-platform/k2-api.h" +#include "runtime-light/stdlib/diagnostics/logs.h" struct ComponentState final : private vk::not_copyable { - ComponentState() noexcept = default; + AllocatorState m_allocator_state{INIT_COMPONENT_ALLOCATOR_SIZE, DEFAULT_MIN_EXTRA_MEMORY_POOL_SIZE, 0}; + kphp::stl::string m_confdata_proxy_actor_name; + +private: + const uint32_t m_argc{k2::args_count()}; + +public: + ComponentState() noexcept; static auto get() noexcept -> const ComponentState&; static auto get_mutable() noexcept -> ComponentState&; + +private: + auto parse_confdata_proxy_actor_name_arg(std::string_view) noexcept -> void; + auto parse_args() noexcept -> void; + + static constexpr std::string_view CONFDATA_PROXY_ACTOR_NAME_ARG{"confdata-proxy-actor-name"}; + static constexpr auto INIT_COMPONENT_ALLOCATOR_SIZE{static_cast(1024U * 1024U)}; // 1MiB }; +inline ComponentState::ComponentState() noexcept { + parse_args(); + + if (m_confdata_proxy_actor_name.empty()) { + kphp::log::error("{} argument is required", CONFDATA_PROXY_ACTOR_NAME_ARG); + } +} + inline auto ComponentState::get() noexcept -> const ComponentState& { return *k2::component_state(); } From 516af38bfaced973a2a1f2785a617ca87dfc4f30 Mon Sep 17 00:00:00 2001 From: Alexander Polyakov Date: Mon, 10 Aug 2026 17:38:09 +0300 Subject: [PATCH 2/9] minor changes --- runtime-light/components/confdata/bindings/bindings.cpp | 9 +++------ .../components/confdata/state/instance-state.cpp | 6 +++--- runtime-light/components/confdata/state/instance-state.h | 8 ++++---- 3 files changed, 10 insertions(+), 13 deletions(-) diff --git a/runtime-light/components/confdata/bindings/bindings.cpp b/runtime-light/components/confdata/bindings/bindings.cpp index 30569fcd5a..9ab3cb6c46 100644 --- a/runtime-light/components/confdata/bindings/bindings.cpp +++ b/runtime-light/components/confdata/bindings/bindings.cpp @@ -19,11 +19,11 @@ namespace kphp::coro { auto instance_state::get() noexcept -> instance_state& { - return InstanceState::get().coroutine_instance_state; + return InstanceState::get().m_coroutine_instance_state; } auto io_scheduler::get() noexcept -> io_scheduler& { - return InstanceState::get().io_scheduler; + return InstanceState::get().m_io_scheduler; } } // namespace kphp::coro @@ -31,9 +31,6 @@ auto io_scheduler::get() noexcept -> io_scheduler& { namespace kphp::log { auto contextual_tags::try_get() noexcept -> std::optional> { - if (k2::instance_state() != nullptr) [[likely]] { - return InstanceState::get().instance_tags; - } return std::nullopt; } @@ -41,7 +38,7 @@ auto contextual_tags::try_get() noexcept -> std::optional const AllocatorState& { if (const auto* instance_state_ptr{k2::instance_state()}; instance_state_ptr != nullptr) [[likely]] { - return instance_state_ptr->instance_allocator_state; + return instance_state_ptr->m_allocator_state; } else if (const auto* component_state_ptr{k2::component_state()}; component_state_ptr != nullptr) { return component_state_ptr->m_allocator_state; } diff --git a/runtime-light/components/confdata/state/instance-state.cpp b/runtime-light/components/confdata/state/instance-state.cpp index 1aaa3c42f4..49e1a94fcb 100644 --- a/runtime-light/components/confdata/state/instance-state.cpp +++ b/runtime-light/components/confdata/state/instance-state.cpp @@ -17,10 +17,10 @@ auto InstanceState::init() noexcept -> void { auto main_task{run()}; // initialize async stack auto& main_task_async_stack_frame{main_task.get_handle().promise().get_async_stack_frame()}; - main_task_async_stack_frame.async_stack_root = std::addressof(coroutine_instance_state.coroutine_stack_root); - coroutine_instance_state.coroutine_stack_root.top_async_stack_frame = std::addressof(main_task_async_stack_frame); + main_task_async_stack_frame.async_stack_root = std::addressof(m_coroutine_instance_state.coroutine_stack_root); + m_coroutine_instance_state.coroutine_stack_root.top_async_stack_frame = std::addressof(main_task_async_stack_frame); // spawn main task onto the scheduler - kphp::log::assertion(io_scheduler.spawn(std::move(main_task))); + kphp::log::assertion(m_io_scheduler.spawn(std::move(main_task))); } auto InstanceState::run() noexcept -> kphp::coro::task<> { diff --git a/runtime-light/components/confdata/state/instance-state.h b/runtime-light/components/confdata/state/instance-state.h index 4a7243baea..84bd1a72e7 100644 --- a/runtime-light/components/confdata/state/instance-state.h +++ b/runtime-light/components/confdata/state/instance-state.h @@ -15,12 +15,12 @@ #include "runtime-light/stdlib/diagnostics/contextual-tags.h" struct InstanceState final : vk::not_copyable { - AllocatorState instance_allocator_state{INIT_INSTANCE_ALLOCATOR_SIZE, DEFAULT_MIN_EXTRA_MEMORY_POOL_SIZE, 0}; + AllocatorState m_allocator_state{INIT_INSTANCE_ALLOCATOR_SIZE, DEFAULT_MIN_EXTRA_MEMORY_POOL_SIZE, 0}; - kphp::log::contextual_tags instance_tags; + kphp::log::contextual_tags m_instance_tags; - kphp::coro::instance_state coroutine_instance_state; - kphp::coro::io_scheduler io_scheduler{coroutine_instance_state}; + kphp::coro::instance_state m_coroutine_instance_state; + kphp::coro::io_scheduler m_io_scheduler{m_coroutine_instance_state}; InstanceState() noexcept = default; static auto get() noexcept -> InstanceState&; From 37a0528a3b405511282e4f72583907fd23575486 Mon Sep 17 00:00:00 2001 From: Alexander Polyakov Date: Mon, 10 Aug 2026 18:44:37 +0300 Subject: [PATCH 3/9] add tl::magic::expect(tl::magic) overload --- runtime-light/tl/tl-types.h | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/runtime-light/tl/tl-types.h b/runtime-light/tl/tl-types.h index 63b4d42a29..609662b69e 100644 --- a/runtime-light/tl/tl-types.h +++ b/runtime-light/tl/tl-types.h @@ -44,6 +44,10 @@ struct magic final { return expected == value; } + bool expect(tl::magic expected) const noexcept { + return expect(expected.value); + } + constexpr size_t footprint() const noexcept { return sizeof(underlying_type); } From ffdd0420fe2c06adddaed98dfda46f4c28c7d555 Mon Sep 17 00:00:00 2001 From: Alexander Polyakov Date: Mon, 10 Aug 2026 18:45:11 +0300 Subject: [PATCH 4/9] [k2] add confdata-proxy TL functions and types --- .../components/confdata/confdata-proxy/tl.h | 136 ++++++++++++++++++ 1 file changed, 136 insertions(+) create mode 100644 runtime-light/components/confdata/confdata-proxy/tl.h diff --git a/runtime-light/components/confdata/confdata-proxy/tl.h b/runtime-light/components/confdata/confdata-proxy/tl.h new file mode 100644 index 0000000000..dfc8fa0082 --- /dev/null +++ b/runtime-light/components/confdata/confdata-proxy/tl.h @@ -0,0 +1,136 @@ +// Compiler for PHP (aka KPHP) +// Copyright (c) 2026 LLC «V Kontakte» +// Distributed under the GPL v3 License, see LICENSE.notice.txt + +#pragma once + +#include +#include +#include +#include +#include + +#include "runtime-light/tl/tl-core.h" +#include "runtime-light/tl/tl-types.h" + +namespace tl::confdata { + +struct keyValuePair final { + tl::string key{}; + tl::string value{}; + tl::Bool is_php_serialized{}; + tl::Bool is_json_serialized{}; + + bool fetch(tl::fetcher& tlf) noexcept { + return key.fetch(tlf) && value.fetch(tlf) && is_php_serialized.fetch(tlf) && is_json_serialized.fetch(tlf); + } + + constexpr size_t footprint() const noexcept { + return key.footprint() + value.footprint() + is_php_serialized.footprint() + is_json_serialized.footprint(); + } +}; + +class KeyValuePair final { + static constexpr tl::magic MAGIC{.value = 0xff1c'b454}; + +public: + tl::confdata::keyValuePair inner{}; + + bool fetch(tl::fetcher& tlf) noexcept { + tl::magic magic{}; + return magic.fetch(tlf) && magic.expect(MAGIC) && inner.fetch(tlf); + } + + constexpr size_t footprint() const noexcept { + return MAGIC.footprint() + inner.footprint(); + } +}; + +struct subscribeResponseOk final { + tl::vector events{}; + tl::string new_page{}; + tl::i64 new_offset{}; + tl::Bool new_has_synced{}; + + bool fetch(tl::fetcher& tlf) noexcept { + return events.fetch(tlf) && new_page.fetch(tlf) && new_offset.fetch(tlf) && new_has_synced.fetch(tlf); + } + + constexpr size_t footprint() const noexcept { + return events.footprint() + new_page.footprint() + new_offset.footprint() + new_has_synced.footprint(); + } +}; + +struct subscribeResponseOldOffsetError final { + bool fetch(tl::fetcher& /*unused*/) noexcept { + return true; + } + + constexpr size_t footprint() const noexcept { + return 0; + } +}; + +class SubscribeResponse final { + static constexpr tl::magic SUBSCRIBE_RESPONSE_OK_MAGIC{.value = 0x2709'63e8}; + static constexpr tl::magic SUBSCRIBE_RESPONSE_OLD_OFFSET_ERROR_MAGIC{.value = 0x11eb'eb02}; + +public: + std::variant value; + + bool fetch(tl::fetcher& tlf) noexcept { + tl::magic magic{}; + if (!magic.fetch(tlf)) { + return false; + } + + if (tl::confdata::subscribeResponseOk response_ok{}; magic.expect(SUBSCRIBE_RESPONSE_OK_MAGIC) && response_ok.fetch(tlf)) { + value.emplace(std::move(response_ok)); + return true; + } + if (tl::confdata::subscribeResponseOldOffsetError old_offset_error{}; + magic.expect(SUBSCRIBE_RESPONSE_OLD_OFFSET_ERROR_MAGIC) && old_offset_error.fetch(tlf)) { + value.emplace(old_offset_error); + return true; + } + return false; + } + + constexpr size_t footprint() const noexcept { + return std::visit( + [](const auto& value) noexcept { + using value_t = std::remove_cvref_t; + if constexpr (std::same_as) { + return SUBSCRIBE_RESPONSE_OK_MAGIC.footprint() + value.footprint(); + } else if constexpr (std::same_as) { + return SUBSCRIBE_RESPONSE_OLD_OFFSET_ERROR_MAGIC.footprint() + value.footprint(); + } else { + static_assert(false, "non-exhaustive visitor!"); + } + }, + value); + } +}; + +class Subscribe final { + static constexpr tl::magic MAGIC{.value = 0xfebd'1230}; + +public: + tl::mask fields_mask{}; + tl::string access_token{}; + tl::string page{}; + tl::i64 offset{}; + tl::Bool has_synced{}; + tl::vector prefixes{}; + + void store(tl::storer& tls) const noexcept { + MAGIC.store(tls), fields_mask.store(tls), access_token.store(tls), page.store(tls), offset.store(tls), has_synced.store(tls), prefixes.store(tls); + } + + constexpr size_t footprint() const noexcept { + return MAGIC.footprint() + fields_mask.footprint() + access_token.footprint() + page.footprint() + offset.footprint() + has_synced.footprint() + + prefixes.footprint(); + } +}; + +} // namespace tl::confdata From 8d68c6e72f1e626e4801810952c21a9ad106a255 Mon Sep 17 00:00:00 2001 From: Alexander Polyakov Date: Wed, 12 Aug 2026 14:28:45 +0300 Subject: [PATCH 5/9] tmp --- .../confdata/confdata-proxy/sync-functions.h | 147 ++++++++++++++++++ .../confdata/state/instance-state.cpp | 80 ++++++++-- .../confdata/state/instance-state.h | 9 ++ runtime-light/stdlib/rpc/rpc-query.h | 68 ++++++++ 4 files changed, 295 insertions(+), 9 deletions(-) create mode 100644 runtime-light/components/confdata/confdata-proxy/sync-functions.h create mode 100644 runtime-light/stdlib/rpc/rpc-query.h diff --git a/runtime-light/components/confdata/confdata-proxy/sync-functions.h b/runtime-light/components/confdata/confdata-proxy/sync-functions.h new file mode 100644 index 0000000000..c0e36dc186 --- /dev/null +++ b/runtime-light/components/confdata/confdata-proxy/sync-functions.h @@ -0,0 +1,147 @@ +// Compiler for PHP (aka KPHP) +// Copyright (c) 2026 LLC «V Kontakte» +// Distributed under the GPL v3 License, see LICENSE.notice.txt + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "common/wrappers/overloaded.h" +#include "runtime-common/core/allocator/script-allocator.h" +#include "runtime-common/core/std/containers.h" +#include "runtime-light/components/confdata/confdata-proxy/tl.h" +#include "runtime-light/coroutine/io-scheduler.h" +#include "runtime-light/coroutine/task.h" +#include "runtime-light/stdlib/diagnostics/logs.h" +#include "runtime-light/stdlib/rpc/rpc-query.h" +#include "runtime-light/tl/tl-core.h" +#include "runtime-light/tl/tl-types.h" + +namespace kphp::confdata { + +struct pagination { + kphp::stl::string m_page; + int64_t m_offset{}; + bool m_has_synced{}; +}; + +enum class subscribe_error : uint8_t { transport, old_offset, malformed_response, not_synced }; + +namespace details { + +// Performs a single confdata.subscribe round-trip. +// On success, invokes `event_handler(events)` once with the batch of received events and updates `to` pagination. +// The batch is a view into the response buffer and is only valid for the duration of the call; empty batches are not delivered. +// An empty event value means that the key has been deleted. +template> event_handler_type> +auto subscribe(std::string_view confdata_proxy_actor, kphp::confdata::pagination& to, + const event_handler_type& event_handler) noexcept -> kphp::coro::task> { + // subscribe is a longpoll method, so the timeout must cover the time confdata-proxy may hold the request open + static constexpr auto SUBSCRIBE_TIMEOUT{std::chrono::seconds{45}}; + + const tl::confdata::Subscribe request{ + .fields_mask = {}, + .access_token = {}, + .page = {.value = to.m_page}, + .offset = {.value = to.m_offset}, + .has_synced = {.value = to.m_has_synced}, + .prefixes = {.value = {{/* a single empty prefix subscribes to all keys */}}}, + }; + tl::storer tls{request.footprint()}; + request.store(tls); + + auto expected_query{kphp::rpc::query::send(confdata_proxy_actor, SUBSCRIBE_TIMEOUT, tls.view())}; + if (!expected_query) [[unlikely]] { + kphp::log::warning("confdata: failed to send subscribe request: {}", expected_query.error()); + co_return std::unexpected{kphp::confdata::subscribe_error::transport}; + } + + kphp::stl::vector response_buffer{}; + auto expected_response{co_await std::move(*expected_query).response([&response_buffer](size_t size) noexcept -> std::span { + response_buffer.resize(size); + return {response_buffer.data(), response_buffer.size()}; + })}; + if (!expected_response) [[unlikely]] { + kphp::log::warning("confdata: failed to fetch subscribe response: {}", expected_response.error()); + co_return std::unexpected{kphp::confdata::subscribe_error::transport}; + } + + tl::fetcher tlf{*expected_response}; + tl::confdata::SubscribeResponse response{}; + if (!response.fetch(tlf)) [[unlikely]] { + kphp::log::warning("confdata: failed to parse subscribe response"); + co_return std::unexpected{kphp::confdata::subscribe_error::malformed_response}; + } + + co_return std::visit( + overloaded{ + [&event_handler, &to](const tl::confdata::subscribeResponseOk& response) noexcept -> std::expected { + if (const auto& events{response.events}; events.size() != 0) { + std::invoke(event_handler, std::span{events.value}); + } + + to.m_page = response.new_page.value; + to.m_offset = response.new_offset.value; + to.m_has_synced = response.new_has_synced.value; + return {}; + }, + [](const tl::confdata::subscribeResponseOldOffsetError& /* unused */) noexcept -> std::expected { + return std::unexpected{kphp::confdata::subscribe_error::old_offset}; + }, + }, + response.value); +} + +} // namespace details + +// Paginates through a consistent snapshot of all subscribed keys until it has been fully synced. +// Returns the final pagination that should be passed to `update`. +// +// `event_handler` is invoked once per round-trip with a batch of events; the batch is only valid +// for the duration of the call and must be copied if it needs to be retained. +template> event_handler_type> +auto sync(std::string_view confdata_proxy_actor, + event_handler_type event_handler) noexcept -> kphp::coro::task> { + kphp::confdata::pagination p{}; + for (; !p.m_has_synced;) { + if (auto expected{co_await details::subscribe(confdata_proxy_actor, p, event_handler)}; !expected) [[unlikely]] { + co_return std::unexpected{expected.error()}; + } + } + co_return std::move(p); +} + +// Longpoll loop: invokes `event_handler` for each event as it arrives, throttled to at most one batch per second: +// events that arrive between round-trips are buffered by the proxy and coalesced into the next batch. Returns only on error; +// `subscribe_error::old_offset` means that the local version is too old and a clean `sync` is required. +// `from` must be a synced pagination, typically the one returned by `sync`; `subscribe_error::not_synced` is returned otherwise. +// +// `event_handler` is invoked once per round-trip with a batch of events; the batch is only valid +// for the duration of the call and must be copied if it needs to be retained. +template> event_handler_type> +auto update(std::string_view confdata_proxy_actor, kphp::confdata::pagination& from, + event_handler_type event_handler) noexcept -> kphp::coro::task> { + // limits the update rate to at most one batch per interval + static constexpr auto UPDATE_INTERVAL{std::chrono::seconds{1}}; + + if (!from.m_has_synced) [[unlikely]] { + co_return std::unexpected{kphp::confdata::subscribe_error::not_synced}; + } + + for (;;) { + if (auto expected{co_await details::subscribe(confdata_proxy_actor, from, event_handler)}; !expected) [[unlikely]] { + co_return std::unexpected{expected.error()}; + } + co_await kphp::coro::io_scheduler::get().schedule(UPDATE_INTERVAL); + } +} + +} // namespace kphp::confdata diff --git a/runtime-light/components/confdata/state/instance-state.cpp b/runtime-light/components/confdata/state/instance-state.cpp index 49e1a94fcb..01510f1bb1 100644 --- a/runtime-light/components/confdata/state/instance-state.cpp +++ b/runtime-light/components/confdata/state/instance-state.cpp @@ -4,15 +4,34 @@ #include "runtime-light/components/confdata/state/instance-state.h" +#include #include +#include #include #include +#include #include +#include "runtime-light/components/confdata/confdata-proxy/sync-functions.h" +#include "runtime-light/components/confdata/confdata-proxy/tl.h" +#include "runtime-light/components/confdata/state/component-state.h" #include "runtime-light/coroutine/task.h" +#include "runtime-light/coroutine/when-all.h" #include "runtime-light/stdlib/diagnostics/logs.h" #include "runtime-light/streams/stream.h" +namespace { + +auto sync_handler(std::span events) noexcept -> void { + kphp::log::info("got {} events on sync", events.size()); +} + +auto update_handler(std::span events) noexcept -> void { + kphp::log::info("got {} events on update", events.size()); +} + +} // namespace + auto InstanceState::init() noexcept -> void { auto main_task{run()}; // initialize async stack @@ -24,16 +43,59 @@ auto InstanceState::init() noexcept -> void { } auto InstanceState::run() noexcept -> kphp::coro::task<> { - auto opt_stream{co_await kphp::component::stream::accept()}; - if (!opt_stream.has_value()) [[unlikely]] { - kphp::log::warning("failed to accept a stream"); - co_return; + co_await kphp::coro::when_all(service_loop(), accept_loop()); // both never return + kphp::log::assertion(false); +} + +auto InstanceState::accept_loop() noexcept -> kphp::coro::task<> { + for (;;) { + auto opt_stream{co_await kphp::component::stream::accept()}; + if (!opt_stream.has_value()) [[unlikely]] { + kphp::log::warning("failed to accept a stream"); + continue; + } + auto request_stream{std::move(*opt_stream)}; + kphp::log::info("accepted a stream: descriptor -> {}", request_stream.descriptor()); + + // dummy implementation: drain the request and close + if (auto expected{co_await request_stream.read_all([](std::span) noexcept {})}; !expected) [[unlikely]] { + kphp::log::warning("failed to read a request: error -> {}", expected.error()); + } } - auto request_stream{std::move(*opt_stream)}; - kphp::log::info("accepted a stream: descriptor -> {}", request_stream.descriptor()); +} + +auto InstanceState::service_loop() noexcept -> kphp::coro::task<> { + static constexpr auto CONFDATA_RETRY_INTERVAL{std::chrono::seconds{1}}; + const std::string_view confdata_proxy_actor{ComponentState::get().m_confdata_proxy_actor_name}; + + for (;;) { + if (!m_pagination.m_has_synced) { + auto sync{co_await kphp::confdata::sync(confdata_proxy_actor, sync_handler)}; + if (!sync) [[unlikely]] { + kphp::log::warning("confdata sync failed: error -> {}, retrying", std::to_underlying(std::move(sync).error())); + co_await m_io_scheduler.schedule(CONFDATA_RETRY_INTERVAL); + continue; + } + m_pagination = *std::move(sync); + m_warmup_status = InstanceState::warmup_status::done; + } - // dummy implementation: drain the request and close - if (auto expected{co_await request_stream.read_all([](std::span) noexcept {})}; !expected) [[unlikely]] { - kphp::log::warning("failed to read a request: error -> {}", expected.error()); + auto update{co_await kphp::confdata::update(confdata_proxy_actor, m_pagination, update_handler)}; + // update returns only on error; m_pagination was advanced in place up to the last applied batch + kphp::log::assertion(!update.has_value()); + switch (update.error()) { + case kphp::confdata::subscribe_error::old_offset: + case kphp::confdata::subscribe_error::not_synced: + // local version is too old: clean re-sync required + kphp::log::warning("confdata update failed: error -> {}, resyncing", std::to_underlying(std::move(update).error())); + m_pagination = {}; + break; + case kphp::confdata::subscribe_error::transport: + case kphp::confdata::subscribe_error::malformed_response: + // pagination is still valid; the longpoll resumes from the last applied position + kphp::log::warning("confdata update failed: error -> {}, retrying", std::to_underlying(std::move(update).error())); + break; + } + co_await m_io_scheduler.schedule(CONFDATA_RETRY_INTERVAL); } } diff --git a/runtime-light/components/confdata/state/instance-state.h b/runtime-light/components/confdata/state/instance-state.h index 84bd1a72e7..8bd4c66434 100644 --- a/runtime-light/components/confdata/state/instance-state.h +++ b/runtime-light/components/confdata/state/instance-state.h @@ -5,9 +5,11 @@ #pragma once #include +#include #include "common/mixin/not_copyable.h" #include "runtime-light/allocator/allocator-state.h" +#include "runtime-light/components/confdata/confdata-proxy/sync-functions.h" #include "runtime-light/coroutine/coroutine-state.h" #include "runtime-light/coroutine/io-scheduler.h" #include "runtime-light/coroutine/task.h" @@ -15,8 +17,13 @@ #include "runtime-light/stdlib/diagnostics/contextual-tags.h" struct InstanceState final : vk::not_copyable { + enum class warmup_status : uint8_t { pending, done }; + AllocatorState m_allocator_state{INIT_INSTANCE_ALLOCATOR_SIZE, DEFAULT_MIN_EXTRA_MEMORY_POOL_SIZE, 0}; + warmup_status m_warmup_status{warmup_status::pending}; + kphp::confdata::pagination m_pagination{}; + kphp::log::contextual_tags m_instance_tags; kphp::coro::instance_state m_coroutine_instance_state; @@ -31,6 +38,8 @@ struct InstanceState final : vk::not_copyable { static constexpr auto INIT_INSTANCE_ALLOCATOR_SIZE = static_cast(16U * 1024U * 1024U); // 16MiB auto run() noexcept -> kphp::coro::task<>; + auto accept_loop() noexcept -> kphp::coro::task<>; + auto service_loop() noexcept -> kphp::coro::task<>; }; inline auto InstanceState::get() noexcept -> InstanceState& { diff --git a/runtime-light/stdlib/rpc/rpc-query.h b/runtime-light/stdlib/rpc/rpc-query.h new file mode 100644 index 0000000000..ef635f3ade --- /dev/null +++ b/runtime-light/stdlib/rpc/rpc-query.h @@ -0,0 +1,68 @@ +// Compiler for PHP (aka KPHP) +// Copyright (c) 2026 LLC «V Kontakte» +// Distributed under the GPL v3 License, see LICENSE.notice.txt + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "common/rpc-error-codes.h" +#include "runtime-light/coroutine/io-scheduler.h" +#include "runtime-light/coroutine/task.h" +#include "runtime-light/k2-platform/k2-api.h" +#include "runtime-light/stdlib/time/time-functions.h" + +namespace kphp::rpc { + +class query { + k2::descriptor m_descriptor{k2::INVALID_PLATFORM_DESCRIPTOR}; + std::chrono::steady_clock::time_point m_deadline; + + query(k2::descriptor descriptor, std::chrono::steady_clock::time_point deadline) noexcept + : m_descriptor{descriptor}, + m_deadline{deadline} {} + + auto drop() noexcept -> void; + +public: + query() = delete; + + query(query&& other) noexcept + : m_descriptor{std::exchange(other.m_descriptor, k2::INVALID_PLATFORM_DESCRIPTOR)}, + m_deadline{other.m_deadline} {} + + query& operator=(query&& other) noexcept { + if (this != std::addressof(other)) { + drop(); + m_descriptor = std::exchange(other.m_descriptor, k2::INVALID_PLATFORM_DESCRIPTOR); + m_deadline = other.m_deadline; + } + return *this; + } + + ~query() { + drop(); + } + + query(const query& other) = delete; + query& operator=(const query& other) = delete; + + static auto send(std::string_view actor, std::chrono::milliseconds timeout, + std::span request_buffer) noexcept -> std::expected; + + template B> + requires std::is_same_v, std::span> + auto response(B response_buffer_provider) && noexcept -> kphp::coro::task, int32_t>>; +}; + +} // namespace kphp::rpc From 0bbbb095721ea0296c5cf8ee114bd02faf44b695 Mon Sep 17 00:00:00 2001 From: Alexander Polyakov Date: Wed, 12 Aug 2026 20:47:47 +0300 Subject: [PATCH 6/9] use custom_timeout_ms --- .../confdata/confdata-proxy/sync-functions.h | 23 +++++++++++-------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/runtime-light/components/confdata/confdata-proxy/sync-functions.h b/runtime-light/components/confdata/confdata-proxy/sync-functions.h index c0e36dc186..c43830286d 100644 --- a/runtime-light/components/confdata/confdata-proxy/sync-functions.h +++ b/runtime-light/components/confdata/confdata-proxy/sync-functions.h @@ -45,16 +45,19 @@ template> event_handl auto subscribe(std::string_view confdata_proxy_actor, kphp::confdata::pagination& to, const event_handler_type& event_handler) noexcept -> kphp::coro::task> { // subscribe is a longpoll method, so the timeout must cover the time confdata-proxy may hold the request open - static constexpr auto SUBSCRIBE_TIMEOUT{std::chrono::seconds{45}}; - - const tl::confdata::Subscribe request{ - .fields_mask = {}, - .access_token = {}, - .page = {.value = to.m_page}, - .offset = {.value = to.m_offset}, - .has_synced = {.value = to.m_has_synced}, - .prefixes = {.value = {{/* a single empty prefix subscribes to all keys */}}}, - }; + static constexpr auto SUBSCRIBE_TIMEOUT{std::chrono::milliseconds{45'000}}; + + const tl::RpcDestActorFlags request{.inner = {.actor_id = {}, + .extra = {.opt_custom_timeout_ms = tl::i32{.value = SUBSCRIBE_TIMEOUT.count()}}, + .query = tl::confdata::Subscribe{ + .fields_mask = {}, + .access_token = {}, + .page = {.value = to.m_page}, + .offset = {.value = to.m_offset}, + .has_synced = {.value = to.m_has_synced}, + .prefixes = {.value = {{/* a single empty prefix subscribes to all keys */}}}, + + }}}; tl::storer tls{request.footprint()}; request.store(tls); From ae1949f0c51c12a82fb213effc1b9119821be333 Mon Sep 17 00:00:00 2001 From: Alexander Polyakov Date: Thu, 13 Aug 2026 13:11:02 +0300 Subject: [PATCH 7/9] make tl::rpcInvokeReqExtra's flags public --- runtime-light/tl/tl-types.h | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/runtime-light/tl/tl-types.h b/runtime-light/tl/tl-types.h index 609662b69e..7fa507edf6 100644 --- a/runtime-light/tl/tl-types.h +++ b/runtime-light/tl/tl-types.h @@ -1287,7 +1287,7 @@ class traceContext final { } // namespace tracing -class rpcInvokeReqExtra final { +struct rpcInvokeReqExtra final { static constexpr uint32_t RETURN_BINLOG_POS_FLAG = vk::tl::common::rpc_invoke_req_extra_flags::return_binlog_pos; static constexpr uint32_t RETURN_BINLOG_TIME_FLAG = vk::tl::common::rpc_invoke_req_extra_flags::return_binlog_time; static constexpr uint32_t RETURN_PID_FLAG = vk::tl::common::rpc_invoke_req_extra_flags::return_pid; @@ -1308,7 +1308,6 @@ class rpcInvokeReqExtra final { static constexpr uint32_t TRACE_CONTEXT_FLAG = vk::tl::common::rpc_invoke_req_extra_flags::trace_context; static constexpr uint32_t EXECUTION_CONTEXT_FLAG = vk::tl::common::rpc_invoke_req_extra_flags::execution_context; -public: bool return_binlog_pos{}; bool return_binlog_time{}; bool return_pid{}; From 2b2c2cd762c70433cccbadd33cd8e79f74dfd312 Mon Sep 17 00:00:00 2001 From: Alexander Polyakov Date: Thu, 13 Aug 2026 13:11:29 +0300 Subject: [PATCH 8/9] fix use of tl::rpcInvokeReqExtra --- .../components/confdata/confdata-proxy/sync-functions.h | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/runtime-light/components/confdata/confdata-proxy/sync-functions.h b/runtime-light/components/confdata/confdata-proxy/sync-functions.h index c43830286d..120a5c8f09 100644 --- a/runtime-light/components/confdata/confdata-proxy/sync-functions.h +++ b/runtime-light/components/confdata/confdata-proxy/sync-functions.h @@ -48,6 +48,7 @@ auto subscribe(std::string_view confdata_proxy_actor, kphp::confdata::pagination static constexpr auto SUBSCRIBE_TIMEOUT{std::chrono::milliseconds{45'000}}; const tl::RpcDestActorFlags request{.inner = {.actor_id = {}, + .flags = {.value = tl::rpcInvokeReqExtra::CUSTOM_TIMEOUT_MS_FLAG}, .extra = {.opt_custom_timeout_ms = tl::i32{.value = SUBSCRIBE_TIMEOUT.count()}}, .query = tl::confdata::Subscribe{ .fields_mask = {}, @@ -61,7 +62,8 @@ auto subscribe(std::string_view confdata_proxy_actor, kphp::confdata::pagination tl::storer tls{request.footprint()}; request.store(tls); - auto expected_query{kphp::rpc::query::send(confdata_proxy_actor, SUBSCRIBE_TIMEOUT, tls.view())}; + // client-side timeout must outlive the server-side longpoll (SUBSCRIBE_TIMEOUT); 10x is a safe margin + auto expected_query{kphp::rpc::query::send(confdata_proxy_actor, SUBSCRIBE_TIMEOUT * 10, tls.view())}; if (!expected_query) [[unlikely]] { kphp::log::warning("confdata: failed to send subscribe request: {}", expected_query.error()); co_return std::unexpected{kphp::confdata::subscribe_error::transport}; From 2ae97080d3361b82c82156d37c0a7f7b22c2fea4 Mon Sep 17 00:00:00 2001 From: Alexander Polyakov Date: Thu, 13 Aug 2026 16:23:00 +0300 Subject: [PATCH 9/9] implement confdata's k2_warmup --- .../components/confdata/confdata-component.cpp | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/runtime-light/components/confdata/confdata-component.cpp b/runtime-light/components/confdata/confdata-component.cpp index d846ce020c..4f7761cd05 100644 --- a/runtime-light/components/confdata/confdata-component.cpp +++ b/runtime-light/components/confdata/confdata-component.cpp @@ -59,7 +59,22 @@ VISIBILITY_DEFAULT void k2_init_instance() { } VISIBILITY_DEFAULT k2::PollStatus k2_warmup() { - return k2::PollStatus::PollFinishedOk; + k2::details::image_state_ptr = k2_image_state(); + k2::details::component_state_ptr = k2_component_state(); + k2::details::instance_state_ptr = k2_instance_state(); + + auto& instance{InstanceState::get()}; + if (instance.m_warmup_status == InstanceState::warmup_status::done) { + return k2::PollStatus::PollFinishedOk; + } + + // the initial sync is performed by the service loop; pump the scheduler and observe the status it sets + const auto poll_status{kphp::coro::io_scheduler::get().process_events()}; + if (instance.m_warmup_status == InstanceState::warmup_status::done) { + return k2::PollStatus::PollFinishedOk; + } + // PollFinishedOk while the sync is still incomplete means the scheduler has drained unexpectedly + return poll_status == k2::PollStatus::PollFinishedOk ? k2::PollStatus::PollFinishedError : poll_status; } VISIBILITY_DEFAULT k2::PollStatus k2_poll() {