[k2] implement confdata synchronization and warmup

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"
@@ -18,29 +19,28 @@
 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
 
 namespace kphp::log {
 
 auto contextual_tags::try_get() noexcept -> std::optional<std::reference_wrapper<contextual_tags>> {
-  if (k2::instance_state() != nullptr) [[likely]] {
-    return InstanceState::get().instance_tags;
-  }
   return std::nullopt;
 }
 
 } // namespace kphp::log
 
 auto AllocatorState::get() noexcept -> 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->m_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");
 }

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() {

runtime-light/components/confdata/confdata-proxy/sync-functions.h

@@ -0,0 +1,153 @@
+// Compiler for PHP (aka KPHP)
+// Copyright (c) 2026 LLC «V Kontakte»
+// Distributed under the GPL v3 License, see LICENSE.notice.txt
+
+#pragma once
+
+#include <chrono>
+#include <concepts>
+#include <cstddef>
+#include <cstdint>
+#include <expected>
+#include <span>
+#include <string_view>
+#include <utility>
+#include <variant>
+
+#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/k2-platform/k2-api.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<kphp::memory::script_allocator> 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<std::invocable<std::span<const tl::confdata::KeyValuePair>> 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<std::expected<void, kphp::confdata::subscribe_error>> {
+  // 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::milliseconds{45'000}};
+
+  const tl::RpcDestActorFlags<tl::confdata::Subscribe> 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 = {},
+                                                                             .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);
+
+  // 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(), k2::RpcKind::TL_RPC)};
+  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<std::byte, kphp::memory::script_allocator> response_buffer{};
+  auto expected_response{co_await kphp::rpc::query::response(std::move(*expected_query), [&response_buffer](size_t size) noexcept -> std::span<std::byte> {
+    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<void, kphp::confdata::subscribe_error> {
+            if (const auto& events{response.events}; events.size() != 0) {
+              std::invoke(event_handler, std::span<const tl::confdata::KeyValuePair>{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<void, kphp::confdata::subscribe_error> {
+            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<std::invocable<std::span<const tl::confdata::KeyValuePair>> event_handler_type>
+auto sync(std::string_view confdata_proxy_actor,
+          event_handler_type event_handler) noexcept -> kphp::coro::task<std::expected<kphp::confdata::pagination, kphp::confdata::subscribe_error>> {
+  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<std::invocable<std::span<const tl::confdata::KeyValuePair>> event_handler_type>
+auto update(std::string_view confdata_proxy_actor, kphp::confdata::pagination& from,
+            event_handler_type event_handler) noexcept -> kphp::coro::task<std::expected<void, kphp::confdata::subscribe_error>> {
+  // 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

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 <concepts>
+#include <cstddef>
+#include <type_traits>
+#include <utility>
+#include <variant>
+
+#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<tl::confdata::KeyValuePair> 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<tl::confdata::subscribeResponseOk, tl::confdata::subscribeResponseOldOffsetError> 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<tl::confdata::subscribeResponseOk>(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<tl::confdata::subscribeResponseOldOffsetError>(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<decltype(value)>;
+          if constexpr (std::same_as<value_t, tl::confdata::subscribeResponseOk>) {
+            return SUBSCRIBE_RESPONSE_OK_MAGIC.footprint() + value.footprint();
+          } else if constexpr (std::same_as<value_t, tl::confdata::subscribeResponseOldOffsetError>) {
+            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<tl::string> 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

runtime-light/components/confdata/confdata.cmake

@@ -3,8 +3,13 @@
 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_TL_SRC
+    ${RUNTIME_LIGHT_DIR}/tl/tl-types.cpp
+    ${RUNTIME_LIGHT_DIR}/tl/tl-functions.cpp)
+
 set(K2_CONFDATA_ALLOCATOR_SRC
     ${RUNTIME_LIGHT_DIR}/allocator/runtime-light-allocator.cpp
     ${RUNTIME_LIGHT_DIR}/memory-resource-impl/monotonic-light-buffer-resource.cpp)
@@ -21,6 +26,7 @@ set(K2_CONFDATA_MEMORY_RESOURCE_SRC
 
 set(K2_CONFDATA_SRC
     ${K2_CONFDATA_COMPONENT_SRC}
+    ${K2_CONFDATA_TL_SRC}
     ${K2_CONFDATA_ALLOCATOR_SRC}
     ${K2_CONFDATA_DIAGNOSTICS_SRC}
     ${K2_CONFDATA_MEMORY_RESOURCE_SRC}