From 2cfc7aa315d5180ea627af6b0db70dae80b3d11d Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Mon, 6 Jul 2026 12:05:39 +0500 Subject: [PATCH 01/10] update serialize variant --- aether/mstream.h | 81 ++++++++++++-- tests/test-types/CMakeLists.txt | 1 + tests/test-types/main.cpp | 2 + tests/test-types/test-variant-type.cpp | 148 +++++++++++++++++++++++++ 4 files changed, 223 insertions(+), 9 deletions(-) create mode 100644 tests/test-types/test-variant-type.cpp diff --git a/aether/mstream.h b/aether/mstream.h index ca412381..4a7d6b64 100644 --- a/aether/mstream.h +++ b/aether/mstream.h @@ -29,24 +29,27 @@ #ifndef AETHER_MSTREAM_H_ #define AETHER_MSTREAM_H_ -#include -#include -#include -#include -#include #include -#include +#include #include #include -#include +#include +#include +#include +#include +#include #include +#include #include #include +#include +#include +#include -#include "aether/clock.h" +#include "aether-miscpp/reflect/domain_visitor.h" // IWYU pragma: keep #include "aether-miscpp/reflect/reflect.h" +#include "aether/clock.h" #include "aether/types/nullable_type.h" -#include "aether-miscpp/reflect/domain_visitor.h" // IWYU pragma: keep namespace ae { @@ -522,6 +525,66 @@ imstream& operator>>(imstream& s, std::optional& v) { return s; } +template +omstream& operator<<(omstream& s, std::variant const& v) { + static_assert(sizeof...(Ts) <= std::numeric_limits::max(), + "std::variant mstream serialization supports at most 255 alternatives (indices 0..254; tag 255 is reserved as the valueless sentinel)"); + if (v.valueless_by_exception()) { + auto const tag = std::numeric_limits::max(); + s << tag; + return s; + } + auto const tag = static_cast(v.index()); + s << tag; + std::visit([&s](auto const& value) { s << value; }, v); + return s; +} + +template +imstream& operator>>(imstream& s, std::variant& v) { + static_assert(sizeof...(Ts) <= std::numeric_limits::max(), + "std::variant mstream serialization supports at most 255 alternatives (indices 0..254; tag 255 is reserved as the valueless sentinel)"); + std::uint8_t tag{}; + s >> tag; + if (!data_was_read(s)) { + return s; + } + + if (tag == std::numeric_limits::max()) { + // Explicit valueless-by-exception sentinel; no payload follows. + return s; + } + + if (tag >= sizeof...(Ts)) { + s.result(ReadResult::kNo); + return s; + } + + auto handled = std::invoke( + [&](std::index_sequence) { + return (std::invoke( + [&](std::integral_constant) { + if (tag == static_cast(I)) { + auto& alternative = v.template emplace(); + s >> alternative; + return true; + } + return false; + }, + std::integral_constant{}) || + ...); + }, + std::make_index_sequence()); + if (!handled) { + s.result(ReadResult::kNo); + return s; + } + if (!data_was_read(s)) { + s.result(ReadResult::kNo); + } + return s; +} + template omstream& operator<<(omstream& s, const std::unique_ptr& v) { if (v) { diff --git a/tests/test-types/CMakeLists.txt b/tests/test-types/CMakeLists.txt index 9ffa42a0..84c30aae 100644 --- a/tests/test-types/CMakeLists.txt +++ b/tests/test-types/CMakeLists.txt @@ -25,6 +25,7 @@ list(APPEND test_srcs test-uid.cpp test-nullable-type.cpp test-address-parser.cpp + test-variant-type.cpp ) if(NOT CM_PLATFORM) diff --git a/tests/test-types/main.cpp b/tests/test-types/main.cpp index 172b5455..58fa592e 100644 --- a/tests/test-types/main.cpp +++ b/tests/test-types/main.cpp @@ -27,6 +27,7 @@ extern int test_static_map(); extern int test_statistics_counter(); extern int test_uid(); extern int test_nullable_type(); +extern int test_variant_type(); extern int test_address_parser(); int main() { @@ -39,6 +40,7 @@ int main() { res += test_statistics_counter(); res += test_uid(); res += test_nullable_type(); + res += test_variant_type(); res += test_address_parser(); return res; } diff --git a/tests/test-types/test-variant-type.cpp b/tests/test-types/test-variant-type.cpp new file mode 100644 index 00000000..de4d51a7 --- /dev/null +++ b/tests/test-types/test-variant-type.cpp @@ -0,0 +1,148 @@ +/* + * Copyright 2026 Aethernet Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include + +#include +#include +#include +#include + +#include "aether/mstream.h" +#include "aether/mstream_buffers.h" + +namespace ae::test_variant_type { +static_assert(std::numeric_limits::max() == 255); +static_assert(std::numeric_limits::max() - 1 == 254); + +// Do not instantiate a 255-alternative std::variant here: on Windows x86 MSVC +// it can explode object sections in this test TU. mstream.h already enforces +// the 255-alternative limit with a static_assert. + +void test_VariantRoundTripExactType() { + std::vector buffer; + std::variant value{std::uint32_t{42}}; + + { + auto buffer_writer = VectorWriter<>{buffer}; + auto stream = omstream{buffer_writer}; + + stream << value; + } + + TEST_ASSERT_EQUAL_UINT8(1, buffer[0]); + + std::variant loaded{}; + { + auto buffer_reader = VectorReader<>{buffer}; + auto stream = imstream{buffer_reader}; + + stream >> loaded; + TEST_ASSERT(data_was_read(stream)); + } + + TEST_ASSERT_EQUAL(1, loaded.index()); + TEST_ASSERT(std::holds_alternative(loaded)); + TEST_ASSERT_EQUAL_UINT32(42, std::get(loaded)); +} + +void test_VariantValuelessSentinel() { + std::vector buffer{std::numeric_limits::max()}; + std::variant value{std::uint32_t{7}}; + + { + auto buffer_reader = VectorReader<>{buffer}; + auto stream = imstream{buffer_reader}; + + stream >> value; + TEST_ASSERT(data_was_read(stream)); + } + + TEST_ASSERT_EQUAL(1, value.index()); + TEST_ASSERT(std::holds_alternative(value)); + TEST_ASSERT_EQUAL_UINT32(7, std::get(value)); +} + +void test_VariantUnsupportedBoundaryTag254Fails() { + std::vector buffer{ + static_cast(std::numeric_limits::max() - 1)}; + std::variant value{std::uint32_t{7}}; + + { + auto buffer_reader = VectorReader<>{buffer}; + auto stream = imstream{buffer_reader}; + + stream >> value; + TEST_ASSERT(!data_was_read(stream)); + } + + TEST_ASSERT_EQUAL(1, value.index()); + TEST_ASSERT(std::holds_alternative(value)); + TEST_ASSERT_EQUAL_UINT32(7, std::get(value)); +} + +void test_VariantBoundaryTag254AndSentinel255() { + std::variant value{ + std::in_place_index<2>, std::uint16_t{9}}; + + std::vector buffer; + { + auto buffer_writer = VectorWriter<>{buffer}; + auto stream = omstream{buffer_writer}; + + stream << value; + } + + TEST_ASSERT_EQUAL_UINT8(2, buffer[0]); + + std::variant loaded{}; + { + auto buffer_reader = VectorReader<>{buffer}; + auto stream = imstream{buffer_reader}; + + stream >> loaded; + TEST_ASSERT(data_was_read(stream)); + } + + TEST_ASSERT_EQUAL(2, loaded.index()); + TEST_ASSERT(std::holds_alternative(loaded)); + TEST_ASSERT_EQUAL_UINT16(9, std::get(loaded)); + + std::vector sentinel_buffer{std::numeric_limits::max()}; + auto sentinel = std::variant{std::in_place_index<0>, 3}; + { + auto buffer_reader = VectorReader<>{sentinel_buffer}; + auto stream = imstream{buffer_reader}; + + stream >> sentinel; + TEST_ASSERT(data_was_read(stream)); + } + + TEST_ASSERT_EQUAL(0, sentinel.index()); + TEST_ASSERT(std::holds_alternative(sentinel)); + TEST_ASSERT_EQUAL_INT(3, std::get(sentinel)); +} + +} // namespace ae::test_variant_type + +int test_variant_type() { + UNITY_BEGIN(); + RUN_TEST(ae::test_variant_type::test_VariantRoundTripExactType); + RUN_TEST(ae::test_variant_type::test_VariantValuelessSentinel); + RUN_TEST(ae::test_variant_type::test_VariantUnsupportedBoundaryTag254Fails); + RUN_TEST(ae::test_variant_type::test_VariantBoundaryTag254AndSentinel255); + return UNITY_END(); +} From 84601685392aa4b5482ef5be69e8990ea1873dfa Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Mon, 6 Jul 2026 12:06:07 +0500 Subject: [PATCH 02/10] fix agents prompts --- opencode.json | 32 ++++++++++++++++++++++++++------ 1 file changed, 26 insertions(+), 6 deletions(-) diff --git a/opencode.json b/opencode.json index 3089c9cb..7cb61a91 100644 --- a/opencode.json +++ b/opencode.json @@ -9,7 +9,7 @@ "model": "openai/gpt-5.5-fast", "reasoningEffort": "medium", "description": "The main agent to rule the others on the way to work on code.", - "prompt": "You are team-manager. You do not write code, build, or test directly. You coordinate specialized agents: @explorer finds relevant files and facts; @architect designs the solution and writes implementation instructions; @coder edits code from precise instructions; @builder validates CMake/Ninja builds and reports compiler or linker errors; @tester runs CTest/unit/smoke tests and reports results; @code-reviewer validates the final diff. Workflow: analyze the request; use @explorer when code context is needed; ask @architect for design and coder checklist; require user approval for medium/high-risk changes; ask @coder to implement; ask @builder to validate the build; if build fails, send the report back to @coder and repeat; ask @tester to run tests; if tests fail, send the report back to @coder and repeat build and tests; ask @code-reviewer for final review; if review blocks, route to @architect or @coder, then rebuild, retest, and review again.", + "prompt": "You are team-manager. You do not write code, build, or test directly. You coordinate specialized agents: @explorer finds relevant files and facts; @architect designs the solution and writes implementation instructions; @coder edits code from precise instructions; @builder validates CMake/Ninja builds and reports compiler or linker errors; @tester runs unit/smoke tests and reports results; @sanity-reviewer performs a fast task/architecture match check; @code-reviewer performs deep final code review. Coordination rule: run workflow stages sequentially. Never start @tester in the same step or same tool batch as @builder. Start @tester only after you have received a successful @builder report for the current implementation. Workflow: analyze the request; use @explorer when code context is needed; ask @architect for design and coder checklist; require user approval for medium/high-risk changes; ask @coder to implement; ask @builder to validate the build; if @builder reports build failure, send the report back to @coder and repeat; after build validation succeeds, ask @tester to run tests; if tests fail, send the report back to @coder and repeat build and tests; after tests pass, ask @sanity-reviewer for a fast task/architecture match check using the changed-file list reported by @coder, and tell @sanity-reviewer to ignore unrelated worktree changes outside that list; if sanity review blocks, route the issue to @coder for implementation mismatch or @architect for architecture mismatch, then rebuild and retest; only after sanity review approves, ask @code-reviewer for deep final review; if deep review blocks, route directly to @architect for revised instructions, then continue with @coder, @builder, @tester, @sanity-reviewer, and @code-reviewer again. Loop control: allow at most three implementation fix cycles for the same task. A fix cycle is @coder -> @builder -> @tester -> @sanity-reviewer -> @code-reviewer. After three failed cycles, or immediately when failures indicate design/API/ownership/lifetime/CMake/requirement mismatch, escalate to @architect to rethink the solution before more coding. If @code-reviewer reports the same issue twice, escalate to @architect. If @coder says the instructions are ambiguous or the fix would change the approved design, escalate to @architect.", "permission": { "edit": "deny", "bash": "deny", @@ -40,7 +40,7 @@ "model": "openai/gpt-5.5", "reasoningEffort": "high", "description": "Analyze requirements and produce C++ architecture and implementation instructions", - "prompt": "You are a solution architect. You do not write code, build, or test. Analyze requirements and existing code facts, preferably using @explorer first when relevant context is missing. Produce practical architecture and precise implementation instructions for @coder. Include: files to inspect or modify, exact API or behavior changes, invariants to preserve, tests or build commands expected, risks, and things not to change. For medium or high-risk changes involving public APIs, persistence, async/task flow, crypto/security, platform behavior, or CMake structure, explicitly request user approval before implementation.", + "prompt": "You are a solution architect. You do not write code, build, or test. Use @explorer first for broad codebase discovery when relevant context is missing. After @explorer reports, read files directly only to verify exact APIs, invariants, ownership/lifetime behavior, or details needed for precise implementation instructions. Avoid repeating broad exploration already completed by @explorer. Produce practical architecture and precise implementation instructions for @coder. Include: files to inspect or modify, exact API or behavior changes, invariants to preserve, tests or build commands expected, risks, and things not to change. For medium or high-risk changes involving public APIs, persistence, async/task flow, crypto/security, platform behavior, or CMake structure, explicitly request user approval before implementation.", "permission": { "edit": "deny", "task": { @@ -59,7 +59,7 @@ "model": "openai/gpt-5.4-mini", "reasoningEffort": "low", "description": "Write c++ code", - "prompt": "You are a focused C++ implementation agent. You receive precise instructions from @team-lead or @architect and implement only those instructions. Follow AGENTS.md, preserve existing style, avoid unrelated refactors, and keep changes minimal. If instructions are ambiguous, ask for clarification instead of inventing architecture. If you need build verification, ask @builder to run it.", + "prompt": "You are a focused C++ implementation agent. You receive precise instructions from @team-lead or @architect and implement only those instructions. Follow AGENTS.md, preserve existing style, avoid unrelated refactors, and keep changes minimal. If instructions are ambiguous, ask for clarification instead of inventing architecture. If a fix requires changing the architect-approved design or you are making a repeated attempt at the same failed issue, stop and ask @architect for revised instructions. Do not run or request build/test validation. After implementation, report what changed and let @team-lead coordinate validation. At the end of your response, report the actual files you changed under: Changed files. Include added, modified, deleted, or renamed files. Do not include files changed by other agents or the user.", "permission": { "edit": "allow", "grep": "allow", @@ -81,7 +81,7 @@ "model": "openai/gpt-5.4-mini-fast", "reasoningEffort": "low", "description": "Validate project build and analyze compiler logs", - "prompt": "You are a C++ build validation specialist. Run the requested CMake/Ninja build: full build, specific target, or configured build command. If the build succeeds, report the command, build directory, and success. If the build fails, analyze the full build log and report root-cause errors only. Collapse cascaded diagnostics into the real underlying issue. Group independent failures by file, target, or symbol. For each issue, report the location, root cause, and brief supporting diagnostic. Do not edit files and do not fix issues yourself.", + "prompt": "You are a C++ build validation specialist. Run the requested CMake/Ninja build: full build, specific target, or configured build command. If the build succeeds, report the command, build directory, and success. If the build fails, analyze the full build log and report root-cause errors only. Collapse cascaded diagnostics into the real underlying issue. Group independent failures by file, target, or symbol. For each issue, report the location, root cause, and brief supporting diagnostic. End every report with exactly one marker: Build validation: SUCCESS or Build validation: FAILURE. Do not edit files and do not fix issues yourself.", "permission": { "edit": "deny", "bash": { @@ -98,9 +98,12 @@ "model": "openai/gpt-5.4-mini-fast", "reasoningEffort": "low", "description": "Run tests and analyze results", - "prompt": "You are a fast test runner and test result reporter. Your task is to run the requested unit tests, CTest tests, or smoke tests and report what passed and what failed. For smoke tests, you may remove build-directory state using the allowed state cleanup command before running a clean smoke test. If a test fails, immediately report the failing test name, command, relevant error output, and a short likely cause. Do not edit files and do not design new tests unless explicitly asked.", + "prompt": "You are a fast test runner and test result reporter. Run tests only after @builder has reported build success for the current implementation. If build success is not confirmed in the current workflow, report that testing is blocked by missing build validation and do not run tests. Your task is to run unit tests and separately run smoke tests, then report what passed and what failed. Before running smoke tests, inspect project instructions such as AGENTS.md to identify what this project defines as smoke tests, where they must be run from, and whether any cleanup is required. If a test fails, report the failing command, relevant output, exit status if available, and a short likely cause. Do not edit files and do not design new tests unless explicitly asked.", "permission": { "edit": "deny", + "read": "allow", + "grep": "allow", + "glob": "allow", "bash": { "*": "deny", "rm -rf *state": "allow", @@ -112,12 +115,29 @@ "temperature": 0.1, "steps": 5 }, + "sanity-reviewer": { + "mode": "subagent", + "model": "openai/gpt-5.4-mini", + "reasoningEffort": "low", + "description": "Fast check that implementation matches the task and proposed architecture", + "prompt": "You are a fast implementation sanity reviewer. Review only the files listed in the changed-file list provided by @coder for the current task. Use the actual diff for those files to verify what changed. Ignore other modified files in the worktree unless they are explicitly included in @coder's changed-file list or explicitly assigned to this task. Check whether the reviewed changes match the user request and architect instructions. Check for missing requested behavior, unrelated changes within the reviewed files, obvious mismatches with the proposed architecture, and incomplete implementation. Do not perform deep C++ review. Do not review generated build artifacts, temporary files, logs, or unrelated files. Never edit files. Report: Reviewed files, Matches task, Matches architecture, Blocking mismatches, Approve or Block.", + "permission": { + "grep": "allow", + "edit": "deny", + "bash": { + "*": "deny", + "git log*": "allow", + "git diff*": "allow" + } + }, + "temperature": 0.1 + }, "code-reviewer": { "mode": "all", "model": "openai/gpt-5.5", "reasoningEffort": "high", "description": "Review code and validate if it solves the problem", - "prompt": "You are a strict C++ code reviewer. Review the current code diff against the user request and architect instructions. Focus on correctness, missing requirements, undefined behavior, object lifetime, ownership, async/task usage, persistence, CMake target propagation, cross-platform desktop/IoT behavior, performance, and security. Do not review generated build artifacts, temporary files, logs, or unrelated files. Never edit files. Report: Findings, Missing tests, Risk assessment, Approve or Block.", + "prompt": "You are a strict C++ code reviewer. Review the current code diff against the user request and architect instructions. Focus on correctness, undefined behavior, object lifetime, ownership, async/task usage, persistence, CMake target propagation, cross-platform desktop/IoT behavior, performance, and security. Do not review generated build artifacts, temporary files, logs, or unrelated files. Never edit files. Mark repeated or design-level issues as Block. When you Block, route the issue to @architect for revised instructions rather than recommending a local coder fix. Report: Findings, Missing tests, Risk assessment, Approve or Block.", "permission": { "grep": "allow", "edit": "deny", From 9ea2ab13cf78b1d39fca407616d46d4b086f0839 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Mon, 6 Jul 2026 12:06:21 +0500 Subject: [PATCH 03/10] make request policy serializable --- aether/cloud_connections/request_policy.h | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/aether/cloud_connections/request_policy.h b/aether/cloud_connections/request_policy.h index 0372b21f..b24e82de 100644 --- a/aether/cloud_connections/request_policy.h +++ b/aether/cloud_connections/request_policy.h @@ -20,23 +20,31 @@ #include #include +#include "aether-miscpp/reflect/reflect.h" + namespace ae { struct RequestPolicy { // Make request only to the main server (the highest priority). - struct MainServer {}; + struct MainServer { + AE_REFLECT() + }; // Make request only to the server with provided priority. struct Priority { - std::size_t priority; //< The less, the higher priority + std::size_t priority{}; //< The less, the higher priority + AE_REFLECT_MEMBERS(priority) }; // Make count of request replicas to the servers. struct Replica { - std::size_t count; + std::size_t count{}; + AE_REFLECT_MEMBERS(count) }; // Make request to all servers. - struct All {}; + struct All { + AE_REFLECT() + }; using Variant = std::variant; }; From d5e30af0e0f8e49e84f7314e295235a69e8f273e Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Mon, 6 Jul 2026 16:24:45 +0500 Subject: [PATCH 04/10] fix assign from empty IActive --- aether/tasks/details/task_subsctiption.h | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/aether/tasks/details/task_subsctiption.h b/aether/tasks/details/task_subsctiption.h index 407a3d4c..81447ec5 100644 --- a/aether/tasks/details/task_subsctiption.h +++ b/aether/tasks/details/task_subsctiption.h @@ -59,7 +59,9 @@ class TaskSubscription final : public ITaskSubscription { TaskSubscription& operator=(IActive* p) noexcept { Reset(); ptr_ = p; - ptr_->active = reinterpret_cast(this); + if (ptr_ != nullptr) { + ptr_->active = reinterpret_cast(this); + } return *this; } From eca44551149faeb8bfad6572e021491c7f4345fe Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Tue, 7 Jul 2026 15:13:31 +0500 Subject: [PATCH 05/10] new ping --- aether/ae_actions/ping.cpp | 177 +++++++++++++++---------------------- aether/ae_actions/ping.h | 47 +++------- 2 files changed, 85 insertions(+), 139 deletions(-) diff --git a/aether/ae_actions/ping.cpp b/aether/ae_actions/ping.cpp index a3a2b860..b53f1029 100644 --- a/aether/ae_actions/ping.cpp +++ b/aether/ae_actions/ping.cpp @@ -17,11 +17,11 @@ #include "aether/ae_actions/ping.h" #if AE_ENABLE_PING -# include +# include # include "aether/server.h" + # include "aether/cloud_connections/cloud_server_connection.h" -# include "aether/server_connections/client_server_connection.h" # include "aether/work_cloud_api/work_server_api/authorized_api.h" # include "aether/ae_actions/ae_actions_tele.h" @@ -29,10 +29,10 @@ namespace ae { Ping::Ping(AeContext const& ae_context, CloudServerConnection& cloud_server_connection, - Duration ping_interval, Duration rx_window, Duration timeout) + Duration next_ping_hint, Duration rx_window, Duration timeout) : ae_context_{ae_context}, cloud_server_connection_{&cloud_server_connection}, - ping_interval_{ping_interval}, + next_ping_hint_{next_ping_hint}, rx_window_{rx_window}, timeout_{timeout}, server_id_{cloud_server_connection_->server()->server_id} { @@ -40,149 +40,114 @@ Ping::Ping(AeContext const& ae_context, kPing, "Ping action created to server id: {}, interval: {:%S}s, rx_window: " "{:%S}s, timeout: {:%S}s", - server_id_, ping_interval_, rx_window_, timeout_); - - ScheduleFirstPing(); + server_id_, next_ping_hint_, rx_window_, timeout_); } Ping::ResultEvent::Subscriber Ping::result_event() { return result_event_; } -void Ping::SetTimeout(Duration timeout) { - // Only next ping will use new timeout - timeout_ = timeout; -} - -void Ping::ScheduleFirstPing() { - // TODO: calculate actual next ping time +void Ping::Start(TimePoint current_time) { auto* cc = cloud_server_connection_->client_connection(); - assert(cc != nullptr && "Client connection is null"); - - // send first ping only after client connection is fully linked - if (cc->stream_info().link_state == LinkState::kLinked) { - // send ping on the next tick - schedule_sub_ = ae_context_.scheduler().Task([&]() { SendPing(); }); - } else { - link_state_sub_ = cc->stream_update_event().Subscribe([this, cc]() { - if (cc->stream_info().link_state == LinkState::kLinked) { - link_state_sub_.Reset(); - // send ping on the next tick - schedule_sub_ = ae_context_.scheduler().Task([&]() { SendPing(); }); - } - }); + assert(cc != nullptr && "Ping::Start requires a client connection"); + assert(cc->stream_info().link_state == LinkState::kLinked && + "Ping::Start requires linked connection"); + assert(!started_ && "Ping::Start must be called only once"); + if (started_) { + return; } -} + started_ = true; -void Ping::SendPing() { AE_TELE_DEBUG(kPingSend, "Send ping"); - auto& write_action = - cloud_server_connection_->client_connection()->AuthorizedApiCall( - SubApi{[this](ApiContext& auth_api) { - auto ping_interval_u64 = static_cast( - std::chrono::duration_cast( - ping_interval_) - .count()); - auto rx_window_u64 = static_cast( - std::chrono::duration_cast( - rx_window_) - .count()); - - auto& pong_promise = - auth_api->ping(ping_interval_u64, rx_window_u64); - auto req_id = pong_promise.request_id(); - - // save the ping request - AE_TELED_DEBUG("Ping server id {}, request {} expected time {:%S}s", - server_id_, req_id, timeout_); - auto current_time = Now(); - auto end_time = current_time + timeout_; - - ping_requests_.push(PingRequest{ - .start = current_time, - .request_id = req_id, - // Wait for response - .wait_result_sub = pong_promise.Subscribe( - [&, req_id](auto&&...) { PingResponse(req_id); }), - .timeout_sub = ae_context_.scheduler().DelayedTask( - [this, req_id]() { PingResponseTimeout(req_id); }, - end_time), - .write_sub = {}, - }); - }}); - - auto& req = ping_requests_.back(); - assert(req.has_value() && - "After call AuthorizedApiCall ping request should be saved"); - - req->write_sub = write_action.status_event().Subscribe([&](auto status) { + auto& write_action = cc->AuthorizedApiCall( + SubApi{[this, current_time](ApiContext& auth_api) { + auto next_ping_hint_ms = static_cast( + std::chrono::duration_cast( + next_ping_hint_) + .count()); + auto rx_window_ms = static_cast( + std::chrono::duration_cast(rx_window_) + .count()); + + auto& pong_promise = auth_api->ping(next_ping_hint_ms, rx_window_ms); + auto req_id = pong_promise.request_id(); + + AE_TELED_DEBUG("Ping server id {}, request {} expected time {:%S}s", + server_id_, req_id, timeout_); + + request_start_ = current_time; + request_id_ = req_id; + + auto wait_result_sub = pong_promise.Subscribe( + [this, req_id](auto&&...) { PingResponse(req_id); }); + wait_result_sub_ = std::move(wait_result_sub); + + auto timeout_sub = ae_context_.scheduler().DelayedTask( + [this, req_id]() { PingResponseTimeout(req_id); }, + current_time + timeout_); + timeout_sub_ = std::move(timeout_sub); + }}); + + write_sub_ = write_action.status_event().Subscribe([this](auto status) { if (status == WriteAction::Status::kFail) { AE_TELE_ERROR(kPingWriteError, "Ping write error"); + ResetRequestSubscriptions(); result_event_.Emit(Error{1}); } }); # if DEBUG - // For debug, call also for get my ip method to print our public ip - // visible to work server - cloud_server_connection_->client_connection()->LoginApiCall( - SubApi{[&](ApiContext& login_api) { - login_api->get_my_ip().Subscribe([sid_ = - server_id_](auto&& res) noexcept { - if (res) { - auto& iip = res.value(); - AE_TELED_DEBUG("Server id: {}, our public ip: {}:{}, coords: {},{}", - sid_, iip.ip, iip.port, iip.latitude, iip.longitude); - } else { - AE_TELED_ERROR("Get my ip failed!"); - } - }); - }}); + cc->LoginApiCall(SubApi{[&](ApiContext& api_call) { + api_call->get_my_ip().Subscribe([&](auto&& res) noexcept { + if (res) { + auto&& ip = std::forward(res).value(); + AE_TELED_DEBUG("Server id: {}, our public ip: {}:{}, coords: {},{}", + server_id_, ip.ip, ip.port, ip.latitude, ip.longitude); + } else { + AE_TELED_ERROR("Get my ip request error {}", + std::forward(res).error()); + } + }); + }}); # endif - - // setup next ping interval - schedule_sub_ = ae_context_.scheduler().DelayedTask([this]() { SendPing(); }, - ping_interval_); } void Ping::PingResponse(RequestId request_id) { - auto request_it = std::find_if( - std::begin(ping_requests_), std::end(ping_requests_), - [&](auto const& p) { return p && (p->request_id == request_id); }); - - if (request_it == std::end(ping_requests_)) { + if (!HasActiveRequest() || request_id_ != request_id) { AE_TELED_WARNING("Got lost, or not our pong response"); return; } - auto& request = *request_it; - auto current_time = Now(); auto ping_duration = - std::chrono::duration_cast(current_time - request->start); - - // reset request as finished - request.reset(); + std::chrono::duration_cast(current_time - request_start_); AE_TELED_DEBUG("Ping server id {} request {} received by {:%S} s", server_id_, request_id, ping_duration); + ResetRequestSubscriptions(); result_event_.Emit(Ok{ping_duration}); } void Ping::PingResponseTimeout(RequestId request_id) { - auto request_it = std::find_if( - std::begin(ping_requests_), std::end(ping_requests_), - [&](auto const& p) { return p && (p->request_id == request_id); }); - - if (request_it == std::end(ping_requests_)) { + if (!HasActiveRequest() || request_id_ != request_id) { AE_TELED_WARNING("Timeout for lost, or not our pong response"); return; } - request_it->reset(); AE_TELE_ERROR(kPingTimeout, "Ping server id {} request {} timeout", server_id_, request_id); + ResetRequestSubscriptions(); result_event_.Emit(Error{2}); } +void Ping::ResetRequestSubscriptions() { + wait_result_sub_.Reset(); + timeout_sub_.Reset(); + write_sub_.Reset(); +} + +bool Ping::HasActiveRequest() const noexcept { + return static_cast(wait_result_sub_); +} + } // namespace ae #endif // AE_ENABLE_PING diff --git a/aether/ae_actions/ping.h b/aether/ae_actions/ping.h index 633a84ba..d94eaaf5 100644 --- a/aether/ae_actions/ping.h +++ b/aether/ae_actions/ping.h @@ -21,73 +21,54 @@ #if AE_ENABLE_PING -# include -# include - -# include "aether/warning_disable.h" - -DISABLE_WARNING_PUSH() -IGNORE_IMPLICIT_CONVERSION() -# include -DISABLE_WARNING_POP() - # include "aether-miscpp/types/result.h" # include "aether/ae_context.h" -# include "aether/events/events.h" -# include "aether/types/server_id.h" # include "aether/api_protocol/request_id.h" # include "aether/events/event_subscription.h" +# include "aether/events/events.h" +# include "aether/tasks/details/task_subsctiption.h" +# include "aether/types/server_id.h" namespace ae { class Channel; class CloudServerConnection; class Ping { - static constexpr std::uint8_t kMaxStorePingTimes = 10; - - struct PingRequest { - TimePoint start; - RequestId request_id; - Subscription wait_result_sub; - TaskSubscription timeout_sub; - Subscription write_sub; - }; - public: using ResultEvent = Event)>; Ping(AeContext const& ae_context, - CloudServerConnection& cloud_server_connection, Duration ping_interval, + CloudServerConnection& cloud_server_connection, Duration next_ping_hint, Duration rx_window, Duration timeout); AE_CLASS_NO_COPY_MOVE(Ping); ResultEvent::Subscriber result_event(); - void SetTimeout(Duration timeout); + void Start(TimePoint current_time); private: - void ScheduleFirstPing(); - void SendPing(); - TimePoint WaitInterval(); - TimePoint WaitResponse(); void PingResponse(RequestId request_id); void PingResponseTimeout(RequestId request_id); + void ResetRequestSubscriptions(); + bool HasActiveRequest() const noexcept; AeContext ae_context_; CloudServerConnection* cloud_server_connection_; - Duration ping_interval_; + Duration next_ping_hint_; Duration rx_window_; Duration timeout_; ServerId server_id_; - etl::circular_buffer, kMaxStorePingTimes> - ping_requests_; + TimePoint request_start_{}; + RequestId request_id_{}; + Subscription wait_result_sub_; + TaskSubscription timeout_sub_; + Subscription write_sub_; ResultEvent result_event_; - Subscription link_state_sub_; - TaskSubscription schedule_sub_; + bool started_{}; }; } // namespace ae #endif // AE_ENABLE_PING From ad08a696c2711e58add83900b44dd9bf3e36bbdc Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Tue, 7 Jul 2026 15:14:36 +0500 Subject: [PATCH 06/10] add client connectivity policy --- aether/CMakeLists.txt | 1 + aether/all.h | 1 + aether/client.cpp | 16 +- aether/client.h | 14 +- aether/client_connectivity_policy.cpp | 157 +++++++++ aether/client_connectivity_policy.h | 154 +++++++++ .../cloud_connections/ping_cloud_servers.cpp | 317 +++++++++++++----- aether/cloud_connections/ping_cloud_servers.h | 70 +++- 8 files changed, 623 insertions(+), 107 deletions(-) create mode 100644 aether/client_connectivity_policy.cpp create mode 100644 aether/client_connectivity_policy.h diff --git a/aether/CMakeLists.txt b/aether/CMakeLists.txt index 65743ba4..22dcda42 100644 --- a/aether/CMakeLists.txt +++ b/aether/CMakeLists.txt @@ -18,6 +18,7 @@ list(APPEND aether_srcs "aether_app.cpp" "aether.cpp" "adapter_registry.cpp" + "client_connectivity_policy.cpp" "client.cpp" "cloud.cpp" "registration_cloud.cpp" diff --git a/aether/all.h b/aether/all.h index 7c39e9d9..1a9a4e1a 100644 --- a/aether/all.h +++ b/aether/all.h @@ -100,6 +100,7 @@ #include "aether/global_ids.h" #include "aether/aether.h" #include "aether/client.h" +#include "aether/client_connectivity_policy.h" #include "aether/server.h" #include "aether/channels/channel.h" #include "aether/crypto.h" diff --git a/aether/client.cpp b/aether/client.cpp index 67a297e6..315fb153 100644 --- a/aether/client.cpp +++ b/aether/client.cpp @@ -50,7 +50,7 @@ ServerKeys* Client::server_state(ServerId server_id) { Cloud::ptr const& Client::cloud() const { return cloud_; } ClientCloudManager::ptr const& Client::cloud_manager() const { - assert(client_cloud_manager_); + assert(client_cloud_manager_.is_valid()); return client_cloud_manager_; } @@ -72,7 +72,8 @@ CloudServerConnections& Client::cloud_connection() { #if AE_ENABLE_PING ping_cloud_servers_ = std::make_unique( - *aether_.Load().as(), *cloud_connection_); + *aether_.Load().as(), *cloud_connection_, + *connectivity_policy().Load()); #endif #if AE_TELE_ENABLED @@ -85,6 +86,11 @@ CloudServerConnections& Client::cloud_connection() { return *cloud_connection_; } +ClientConnectivityPolicy::ptr const& Client::connectivity_policy() { + assert(connectivity_policy_.is_valid()); + return connectivity_policy_; +} + P2pMessageStreamManager& Client::message_stream_manager() { if (!message_stream_manager_) { message_stream_manager_ = std::make_unique( @@ -106,8 +112,12 @@ void Client::SetConfig(std::string client_id, Uid parent_uid, Uid uid, server_keys_.emplace(s->server_id, ServerKeys{s->server_id, master_key_}); } + connectivity_policy_ = ClientConnectivityPolicy::ptr::Create( + CreateWith{domain}.with_flags(ObjFlags::kUnloadedByDefault)); + client_cloud_manager_ = ClientCloudManager::ptr::Create( - domain, Aether::ptr{aether_}, Client::ptr::MakeFromThis(this)); + CreateWith{domain}.with_flags(ObjFlags::kUnloadedByDefault), + Aether::ptr{aether_}, Client::ptr::MakeFromThis(this)); } void Client::SendTelemetry() { diff --git a/aether/client.h b/aether/client.h index 9090a556..4a34d34f 100644 --- a/aether/client.h +++ b/aether/client.h @@ -17,18 +17,19 @@ #ifndef AETHER_CLIENT_H_ #define AETHER_CLIENT_H_ +#include #include #include -#include -#include "aether/memory.h" +#include "aether/client_connectivity_policy.h" #include "aether/cloud.h" +#include "aether/memory.h" #include "aether/obj/obj.h" -#include "aether/types/uid.h" #include "aether/server_keys.h" +#include "aether/types/uid.h" -#include "aether/cloud_connections/ping_cloud_servers.h" #include "aether/cloud_connections/cloud_server_connections.h" +#include "aether/cloud_connections/ping_cloud_servers.h" #include "aether/connection_manager/client_cloud_manager.h" #include "aether/connection_manager/server_connection_manager.h" @@ -61,6 +62,7 @@ class Client : public Obj { ClientCloudManager::ptr const& cloud_manager() const; ServerConnectionManager& server_connection_manager(); CloudServerConnections& cloud_connection(); + ClientConnectivityPolicy::ptr const& connectivity_policy(); P2pMessageStreamManager& message_stream_manager(); void SetConfig(std::string client_id, Uid parent_uid, Uid uid, @@ -68,8 +70,7 @@ class Client : public Obj { AE_OBJECT_REFLECT(AE_MMBRS(aether_, client_id_, parent_uid_, uid_, ephemeral_uid_, master_key_, cloud_, server_keys_, - client_cloud_manager_)) - + connectivity_policy_, client_cloud_manager_)) void SendTelemetry(); private: @@ -85,6 +86,7 @@ class Client : public Obj { // states std::map server_keys_; + ClientConnectivityPolicy::ptr connectivity_policy_; ClientCloudManager::ptr client_cloud_manager_; std::unique_ptr server_connection_manager_; std::unique_ptr cloud_connection_; diff --git a/aether/client_connectivity_policy.cpp b/aether/client_connectivity_policy.cpp new file mode 100644 index 00000000..7b0ce875 --- /dev/null +++ b/aether/client_connectivity_policy.cpp @@ -0,0 +1,157 @@ +/* + * Copyright 2026 Aethernet Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "aether/client_connectivity_policy.h" + +#include + +namespace ae { + +namespace { +constexpr auto kDefaultTiming = RxTiming{ + .conf = RxTimingConf::Every(std::chrono::milliseconds{AE_PING_INTERVAL_MS}), + .next_rx_point = {}, + .recordet_at = {}}; +; + +std::array MakeDefaultRxTimings() { + std::array timings{}; + timings.fill(kDefaultTiming); + return timings; +} +} // namespace + +ClientConnectivityPolicy::RxTimingConfig::RxTimingConfig( + ClientConnectivityPolicy& policy, RequestPolicy::Variant targets) + : policy_{&policy} { + policy_->rx_targets_ = std::move(targets); +} + +ClientConnectivityPolicy::RxTimingConfig& +ClientConnectivityPolicy::RxTimingConfig::ForAllPriorities(RxTimingConf conf) { + for (auto& item : policy_->rx_timings_) { + item.conf = conf; + } + return *this; +} + +ClientConnectivityPolicy::SuspendBlocker::SuspendBlocker( + ClientConnectivityPolicy& policy) + : policy_{&policy} { + policy_->IncrementSuspendBlock(); +} + +ClientConnectivityPolicy::SuspendBlocker::~SuspendBlocker() { Reset(); } + +ClientConnectivityPolicy::SuspendBlocker::SuspendBlocker( + SuspendBlocker&& other) noexcept + : policy_{std::exchange(other.policy_, nullptr)} {} + +ClientConnectivityPolicy::SuspendBlocker& +ClientConnectivityPolicy::SuspendBlocker::operator=( + SuspendBlocker&& other) noexcept { + if (this != &other) { + Reset(); + policy_ = std::exchange(other.policy_, nullptr); + } + return *this; +} + +void ClientConnectivityPolicy::SuspendBlocker::Reset() { + if (policy_ != nullptr) { + policy_->DecrementSuspendBlock(); + policy_ = nullptr; + } +} + +ClientConnectivityPolicy::ClientConnectivityPolicy() + : rx_targets_{RequestPolicy::All{}}, rx_timings_{MakeDefaultRxTimings()} {} + +#ifdef AE_DISTILLATION +ClientConnectivityPolicy::ClientConnectivityPolicy(ObjProp prop) + : Base{prop}, + rx_targets_{RequestPolicy::All{}}, + rx_timings_{MakeDefaultRxTimings()} {} +#endif + +auto ClientConnectivityPolicy::ConfigureRxTimings( + RequestPolicy::Variant targets) -> RxTimingConfig { + return RxTimingConfig{*this, std::move(targets)}; +} + +ClientConnectivityPolicy::SuspendBlocker +ClientConnectivityPolicy::AcquireSuspendBlock() { + return SuspendBlocker{*this}; +} + +ConnectivityStatus ClientConnectivityPolicy::GetStatus() const noexcept { + auto current_time = Now(); + auto next_service_time = TimePoint::max(); + for (auto const& t : rx_timings_) { + next_service_time = std::min( + next_service_time, + (t.recordet_at > current_time) ? current_time : t.next_rx_point); + } + return ConnectivityStatus{.can_suspend = can_suspend_, + .suspend_block_count = suspend_block_count_, + .next_service_time = next_service_time}; +} + +void ClientConnectivityPolicy::ResetRxTimings() { + for (auto& t : rx_timings_) { + t.next_rx_point = {}; + t.recordet_at = {}; + } +} + +void ClientConnectivityPolicy::ReportNextServiceTime( + std::size_t priority, TimePoint next_service_time) { + assert(priority < rx_timings_.size() && "Invalid priority value"); + + auto& t = rx_timings_.at(priority); + t.next_rx_point = next_service_time; + t.recordet_at = Now(); +} + +void ClientConnectivityPolicy::ResetRuntimeState() { + auto current_time = Now(); + for (auto& t : rx_timings_) { + // if clock was reset, also reset next rx points + if (current_time < t.recordet_at) { + t.next_rx_point = {}; + t.recordet_at = {}; + } + } +} + +void ClientConnectivityPolicy::IncrementSuspendBlock() { + ++suspend_block_count_; + can_suspend_ = false; +} + +void ClientConnectivityPolicy::DecrementSuspendBlock() { + assert(suspend_block_count_ > 0); + if (suspend_block_count_ == 0) { + return; + } + --suspend_block_count_; + can_suspend_ = suspend_block_count_ == 0; + if (can_suspend_) { + suspend_allowed_event_.Emit(); + } +} + +} // namespace ae diff --git a/aether/client_connectivity_policy.h b/aether/client_connectivity_policy.h new file mode 100644 index 00000000..e8265c40 --- /dev/null +++ b/aether/client_connectivity_policy.h @@ -0,0 +1,154 @@ +/* + * Copyright 2026 Aethernet Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef AETHER_CLIENT_CONNECTIVITY_POLICY_H_ +#define AETHER_CLIENT_CONNECTIVITY_POLICY_H_ + +#include +#include +#include +#include + +#include "aether/config.h" +#include "aether/events/events.h" +#include "aether/obj/obj.h" + +#include "aether/cloud_connections/request_policy.h" + +namespace ae { + +inline constexpr std::size_t kMaxRxServerPriorities{ + AE_CLOUD_MAX_SERVER_CONNECTIONS}; +static_assert(kMaxRxServerPriorities > 0); + +struct RxTimingConf { + AE_REFLECT_MEMBERS(interval, rx_window) + + Duration interval{}; + Duration rx_window{}; + + static constexpr RxTimingConf Every(Duration i) { + return RxTimingConf{.interval = i, .rx_window = i}; + } + + constexpr RxTimingConf WithWindow(Duration rx_w) const { + return RxTimingConf{.interval = interval, .rx_window = rx_w}; + } +}; + +struct RxTiming { + AE_REFLECT_MEMBERS(conf, next_rx_point, recordet_at) + + RxTimingConf conf; + TimePoint next_rx_point; + TimePoint recordet_at; +}; + +struct ConnectivityStatus { + bool can_suspend{true}; + std::uint8_t suspend_block_count{}; + TimePoint next_service_time; +}; + +class ClientConnectivityPolicy : public Obj { + AE_OBJECT(ClientConnectivityPolicy, Obj, 0) + + public: + class RxTimingConfig { + public: + RxTimingConfig(ClientConnectivityPolicy& policy, + RequestPolicy::Variant targets); + + RxTimingConfig& ForAllPriorities(RxTimingConf conf); + template + RxTimingConfig& ForPriority(RxTimingConf conf) { + static_assert(Priority < kMaxRxServerPriorities); + policy_->rx_timings_[Priority].conf = conf; + return *this; + } + + private: + ClientConnectivityPolicy* policy_; + }; + + class SuspendBlocker { + public: + SuspendBlocker() = default; + explicit SuspendBlocker(ClientConnectivityPolicy& policy); + ~SuspendBlocker(); + + SuspendBlocker(SuspendBlocker&& other) noexcept; + SuspendBlocker& operator=(SuspendBlocker&& other) noexcept; + SuspendBlocker(SuspendBlocker const&) = delete; + SuspendBlocker& operator=(SuspendBlocker const&) = delete; + + void Reset(); + + private: + ClientConnectivityPolicy* policy_{}; + }; + + ClientConnectivityPolicy(); +#ifdef AE_DISTILLATION + explicit ClientConnectivityPolicy(ObjProp prop); +#endif + + AE_CLASS_NO_COPY_MOVE(ClientConnectivityPolicy); + + AE_OBJECT_REFLECT(AE_MMBRS(rx_targets_, rx_timings_)) + template + void Load(CurrentVersion, Dnv& dnv) { + dnv(base_, rx_targets_, rx_timings_); + ResetRuntimeState(); + } + + RxTimingConfig ConfigureRxTimings( + RequestPolicy::Variant targets = RequestPolicy::All{}); + + RequestPolicy::Variant const& rx_targets() const noexcept { + return rx_targets_; + } + std::array const& rx_timings() + const noexcept { + return rx_timings_; + } + Event* suspend_allowed_event() noexcept { + return &suspend_allowed_event_; + } + + ConnectivityStatus GetStatus() const noexcept; + void ResetRxTimings(); + + SuspendBlocker AcquireSuspendBlock(); + void ReportNextServiceTime(std::size_t priority, TimePoint next_service_time); + + private: + void ResetRuntimeState(); + void IncrementSuspendBlock(); + void DecrementSuspendBlock(); + + RequestPolicy::Variant rx_targets_; + std::array rx_timings_; + + bool can_suspend_{true}; + std::uint8_t suspend_block_count_{}; + + Event suspend_allowed_event_; +}; + +} // namespace ae + +#endif // AETHER_CLIENT_CONNECTIVITY_POLICY_H_ diff --git a/aether/cloud_connections/ping_cloud_servers.cpp b/aether/cloud_connections/ping_cloud_servers.cpp index a04aa9ab..61480d8a 100644 --- a/aether/cloud_connections/ping_cloud_servers.cpp +++ b/aether/cloud_connections/ping_cloud_servers.cpp @@ -16,19 +16,219 @@ #include "aether/cloud_connections/ping_cloud_servers.h" +#include +#include + #if AE_ENABLE_PING -# include "aether/server.h" # include "aether/channels/channel.h" +# include "aether/executors/executors.h" +# include "aether/server.h" # include "aether/cloud_connections/cloud_connections_tele.h" namespace ae { + +PingCloudServers::ServerPing::ServerPing(AeContext const& ae_context, + ClientConnectivityPolicy& policy, + CloudServerConnection& cloud_sc, + std::size_t priority) + : ae_context_{ae_context}, + policy_{&policy}, + cloud_sc_{&cloud_sc}, + priority_{priority} { + assert(priority < policy_->rx_timings().size() && + "Server ping priority should be in timings range"); + + auto const& timings = policy_->rx_timings()[priority_]; + timing_conf_ = timings.conf; + + // if it's to early for next rx wait a bit + if ((timings.next_rx_point != TimePoint{}) && + (Now() < timings.next_rx_point)) { + AE_TELED_DEBUG("Wait a bit for next rx point till {:%H:%M:%S}", + timings.next_rx_point); + start_sub_ = ae_context_.scheduler().DelayedTask([&]() { Start(); }, + timings.next_rx_point); + } else { + // acquire suspend for first ping + ping_blocker_ = policy_->AcquireSuspendBlock(); + start_sub_ = ae_context_.scheduler().Task([&]() { Start(); }); + } +} + +PingCloudServers::ServerPing::~ServerPing() = default; + +void PingCloudServers::ServerPing::Stop() { + stop_ = true; + + start_sub_.Reset(); + rx_window_sub_.Reset(); + restream_sub_.Reset(); + + ping_blocker_.Reset(); + rx_window_blocker_.Reset(); + restream_blocker_.Reset(); +} + +template +void PingCloudServers::ServerPing::WaitForLink(ClientServerConnection& cc, + F&& f) { + link_state_sub_ = cc.stream_update_event().Subscribe( + [this, f_ = std::forward(f)]() noexcept { + auto* cc = cloud_sc_->client_connection(); + if (cc->stream_info().link_state == LinkState::kLinked) { + link_state_sub_.Reset(); + std::invoke(f_); + } + }); +} + +auto PingCloudServers::ServerPing::EnsureLinked() { + return ex::create( + [&](auto& ctx) noexcept { + auto* cc = cloud_sc_->client_connection(); + if (cc == nullptr) { + return ex::set_error(std::move(ctx.receiver), 1); + } + if (cc->stream_info().link_state == LinkState::kLinked) { + return ex::set_value(std::move(ctx.receiver)); + } + WaitForLink(*cc, [&]() noexcept { + return ex::set_value(std::move(ctx.receiver)); + }); + }); +} + +auto PingCloudServers::ServerPing::MakePing() { + return ex::let_value([&]() noexcept { + return ex::create( + [&](auto& ctx) noexcept { + // make ping action with timeout based on response statistics + // and timing properties for current server RxTimings + // during the ping and rx window setup suspend blocker + // and save expected next_ping_time_ + auto* cc = cloud_sc_->client_connection(); + assert(cc != nullptr && "Client connection should exists"); + + auto c = cc->server_connection().current_channel(); + if (c == nullptr) { + AE_TELED_ERROR("Current channel value invalid"); + return ex::set_error(std::move(ctx.receiver), 2); + } + + ping_.emplace(ae_context_, *cloud_sc_, timing_conf_.interval, + timing_conf_.rx_window, c->ResponseTimeout()); + + ping_blocker_ = policy_->AcquireSuspendBlock(); + ping_->result_event().Subscribe([this](auto const& res) noexcept { + OnPingResult(res); + ping_blocker_.Reset(); + }); + + // run ping request and open rx window + auto const current_time = Now(); + ping_->Start(current_time); + OpenRxWindow(current_time); + next_ping_time_ = current_time + timing_conf_.interval; + policy_->ReportNextServiceTime(priority_, next_ping_time_); + AE_TELED_DEBUG( + "Next ping time for priority {} at {:%H:%M:%S} after {:%S}", + next_ping_time_, timing_conf_.interval); + + return ex::set_value(std::move(ctx.receiver)); + }); + }); +} + +void PingCloudServers::ServerPing::Start() { + waiter_.emplace( + ae_context_, EnsureLinked() | + ex::let_value([&]() noexcept + -> ex::variant_sender { + if (stop_) { + return ex::just_stopped(); + } + return ex::just(); + }) | + MakePing() | + // track Stop command + ex::let_value( + [&]() noexcept + -> ex::variant_sender { + if (stop_) { + return ex::just_stopped(); + } + return ex::just(); + }), + [this](std::optional&& res) noexcept { + if (res && res->IsOk()) { + // repeat start on next_ping_time_ + start_sub_ = ae_context_.scheduler().DelayedTask( + [&]() noexcept { Start(); }, // ~['_']~ + next_ping_time_); + } else if (res && res->IsErr()) { + AE_TELED_ERROR("Ping start error {}", std::move(res)->error()); + } else { + AE_TELED_DEBUG("Server ping stopped"); + } + }); +} + +void PingCloudServers::ServerPing::OnPingResult( + Result const& res) { + auto* cc = cloud_sc_->client_connection(); + if (cc == nullptr) { + AE_TELED_ERROR("Client connection is null"); + return; + } + + if (res) { + auto c = cc->server_connection().current_channel(); + if (!c) { + AE_TELED_ERROR("Ping's channel value invalid"); + } else { + c->channel_statistics().AddResponseTime(res.value()); + } + } else { + AE_TELED_ERROR("Ping error!"); + ScheduleRestream(); + } +} + +void PingCloudServers::ServerPing::OpenRxWindow(TimePoint sent_time) { + // keep rx window suspend block for timing_.rx_window time + rx_window_blocker_ = policy_->AcquireSuspendBlock(); + rx_window_sub_ = ae_context_.scheduler().DelayedTask( + [this]() { rx_window_blocker_.Reset(); }, + sent_time + timing_conf_.rx_window); +} + +void PingCloudServers::ServerPing::ScheduleRestream() { + if (stop_) { + return; + } + + // TODO: should we block till restream? + restream_blocker_ = policy_->AcquireSuspendBlock(); + restream_sub_ = ae_context_.scheduler().Task([this]() { + auto* cc = cloud_sc_->client_connection(); + if (cc != nullptr) { + cc->Restream(); + } + restream_blocker_.Reset(); + }); +} + PingCloudServers::PingCloudServers( AeContext const& ae_context, - CloudServerConnections& cloud_server_connections) + CloudServerConnections& cloud_server_connections, + ClientConnectivityPolicy& policy) : ae_context_{ae_context}, cloud_server_connections_{&cloud_server_connections}, + policy_{&policy}, servers_update_{ cloud_server_connections_->servers_update_event().Subscribe( MethodPtr<&PingCloudServers::ServersUpdate>{this})} { @@ -36,112 +236,57 @@ PingCloudServers::PingCloudServers( ServersUpdate(); } +PingCloudServers::~PingCloudServers() { task_sub_.Reset(); } + void PingCloudServers::ServersUpdate() { AE_TELED_DEBUG("Servers update"); - // enqueue redispatch for servers if (!task_sub_) { AE_TELED_DEBUG("Enqueue dispatch servers"); - task_sub_ = ae_context_.scheduler().Task([this]() { DispatchToServers(); }); + auto blocker = policy_->AcquireSuspendBlock(); + task_sub_ = ae_context_.scheduler().Task( + [this, blocker = std::move(blocker)]() mutable { + DispatchToServers(); + }); } } void PingCloudServers::DispatchToServers() { + std::set visited_ids; cloud_server_connections_->ForServers( - [this](CloudServerConnection* cloud_sc) { + [this, &visited_ids](CloudServerConnection* cloud_sc) { if (cloud_sc == nullptr) { AE_TELED_ERROR("Visit empty cloud server connection!"); return; } - MakePingToServer(*cloud_sc); + auto server = cloud_sc->server(); + if (server) { + visited_ids.insert(server->server_id); + ReconcileServer(server, *cloud_sc); + } }, - // TODO: config request policy - RequestPolicy::All{}); -} - -void PingCloudServers::MakePingToServer(CloudServerConnection& cloud_sc) { - auto [sid, server_ping, is_new] = GetOrCreatePing(cloud_sc); - AE_TELED_DEBUG("Make ping to server {}, is_new {}", sid, is_new); - if (!is_new) { - return; - } + policy_->rx_targets()); - auto c = cloud_sc.client_connection()->server_connection().current_channel(); - if (!c) { - AE_TELED_ERROR("Current channel value invalid"); - return; - } - auto timeout = c->ResponseTimeout(); - constexpr auto kDefaultInterval = - std::chrono::milliseconds{AE_PING_INTERVAL_MS}; - constexpr auto kDefaultRxWindow = kDefaultInterval; - - // TODO: different ping interval depends on priority - server_ping.ping = std::make_unique( - ae_context_, cloud_sc, kDefaultInterval, kDefaultRxWindow, timeout); - - // subscribe to ping result - // if success get ping time from result and save for statistics - // if failed request Restream for what server connection - server_ping.ping->result_event().Subscribe([this, sid_ = sid, - csc_ = &cloud_sc](auto res) { - auto it = server_pings_.find(sid_); - if (it == server_pings_.end()) { - return; + // stop pings for servers not in connection anymore + for (auto& [sid, sp] : server_pings_) { + if (!visited_ids.contains(sid)) { + sp->Stop(); } - auto& sp = it->second; - if (res) { - // successful ping save timeout to the current channel - auto c = csc_->client_connection()->server_connection().current_channel(); - if (!c) { - AE_TELED_ERROR("Ping's channel value invalid"); - return; - } - c->channel_statistics().AddResponseTime(res.value()); - // and update timeout value - sp.ping->SetTimeout(c->ResponseTimeout()); - } else { - // ping error - AE_TELED_ERROR("Ping error!"); - // after restream if server is changed - csc_->Restream(); - } - }); - // reset ping on server error - server_ping.server_state_sub = cloud_sc.client_connection() - ->server_connection() - .server_error_event() - .Subscribe([this, sid_ = sid]() { - auto it = server_pings_.find(sid_); - if (it == server_pings_.end()) { - return; - } - it->second.ping.reset(); - }); + } } -auto PingCloudServers::GetOrCreatePing(CloudServerConnection& cloud_sc) - -> std::tuple { - auto server = cloud_sc.server(); - assert(server && "Server must exists"); +void PingCloudServers::ReconcileServer(Ptr const& server, + CloudServerConnection& cloud_sc) { + auto const server_id = server->server_id; + auto const priority = cloud_sc.priority(); - auto server_id = server->server_id; auto it = server_pings_.find(server_id); - // Create new ping or replace ping if server's priority changed - if ((it == server_pings_.end()) || - (it->second.priority != cloud_sc.priority())) { - auto [new_it, _] = server_pings_.insert_or_assign( - server_id, ServerPing{ - .ping{}, - .priority = cloud_sc.priority(), - .server_state_sub = {}, - }); - return {server_id, new_it->second, true}; + if ((it == server_pings_.end()) || (it->second->priority() != priority)) { + server_pings_.insert_or_assign( + server_id, std::make_unique(ae_context_, *policy_, cloud_sc, + priority)); + return; } - - // if ping not has value return is_new - return {server_id, it->second, !it->second.ping}; } - } // namespace ae #endif diff --git a/aether/cloud_connections/ping_cloud_servers.h b/aether/cloud_connections/ping_cloud_servers.h index 76738b75..97b19970 100644 --- a/aether/cloud_connections/ping_cloud_servers.h +++ b/aether/cloud_connections/ping_cloud_servers.h @@ -22,42 +22,88 @@ # include # include -# include +# include # include "aether/ae_context.h" -# include "aether/types/server_id.h" # include "aether/events/event_subscription.h" +# include "aether/executors/executors.h" # include "aether/tasks/manual_task_scheduler.h" -# include "aether/cloud_connections/cloud_server_connections.h" +# include "aether/types/server_id.h" # include "aether/ae_actions/ping.h" +# include "aether/client_connectivity_policy.h" +# include "aether/cloud_connections/cloud_server_connections.h" +# include "aether/server.h" namespace ae { class PingCloudServers { - struct ServerPing { - std::unique_ptr ping; - std::size_t priority; - Subscription server_state_sub; + class ServerPing { + public: + ServerPing(AeContext const& ae_context, ClientConnectivityPolicy& policy, + CloudServerConnection& cloud_sc, std::size_t priority); + ~ServerPing(); + + AE_CLASS_NO_COPY_MOVE(ServerPing) + + void Stop(); + + TimePoint next_service_time() const noexcept { return next_ping_time_; } + std::size_t priority() const noexcept { return priority_; } + RxTimingConf const& timing() const noexcept { return timing_conf_; } + + private: + void Start(); + + template + void WaitForLink(ClientServerConnection& cc, F&& f); + + auto EnsureLinked(); + auto MakePing(); + + void OnPingResult(Result const& res); + void OpenRxWindow(TimePoint sent_time); + void ScheduleRestream(); + + AeContext ae_context_; + ClientConnectivityPolicy* policy_; + CloudServerConnection* cloud_sc_; + RxTimingConf timing_conf_{}; + std::size_t priority_{}; + + std::optional> + waiter_; + std::optional ping_; + bool stop_{false}; + Subscription link_state_sub_; + TaskSubscription start_sub_; + TaskSubscription rx_window_sub_; + TaskSubscription restream_sub_; + ClientConnectivityPolicy::SuspendBlocker ping_blocker_; + ClientConnectivityPolicy::SuspendBlocker rx_window_blocker_; + ClientConnectivityPolicy::SuspendBlocker restream_blocker_; + TimePoint next_ping_time_; }; public: PingCloudServers(AeContext const& ae_context, - CloudServerConnections& cloud_server_connections); + CloudServerConnections& cloud_server_connections, + ClientConnectivityPolicy& policy); + ~PingCloudServers(); private: void ServersUpdate(); void DispatchToServers(); - void MakePingToServer(CloudServerConnection& cloud_sc); - std::tuple GetOrCreatePing( - CloudServerConnection& cloud_sc); + void ReconcileServer(Ptr const& server, + CloudServerConnection& cloud_sc); AeContext ae_context_; CloudServerConnections* cloud_server_connections_; + ClientConnectivityPolicy* policy_; Subscription servers_update_; TaskSubscription task_sub_; - std::map server_pings_; + std::map> server_pings_; }; } // namespace ae From 5efd6746a1ab87e93f8d4c758b4f3d2d93cb5c80 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Tue, 7 Jul 2026 15:14:47 +0500 Subject: [PATCH 07/10] make clang-format sort includes --- .clang-format | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/.clang-format b/.clang-format index 980ccab3..90bc40ed 100644 --- a/.clang-format +++ b/.clang-format @@ -1,8 +1,7 @@ --- -Language: Cpp -BasedOnStyle: Google +Language: Cpp +BasedOnStyle: Google IncludeBlocks: Preserve -SortIncludes: Never IndentPPDirectives: AfterHash InsertNewlineAtEOF: On From 9e105d5da55b6f59be9aa5d9f0bd9f352c5fc5ee Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Tue, 7 Jul 2026 16:05:10 +0500 Subject: [PATCH 08/10] remove uap and time sync --- aether/CMakeLists.txt | 8 +- aether/ae_actions/time_sync.cpp | 246 ------------------ aether/ae_actions/time_sync.h | 87 ------- aether/aether.cpp | 20 -- aether/aether.h | 19 +- aether/aether_app.cpp | 16 -- aether/aether_app.h | 9 - aether/all.h | 1 - aether/clock.h | 45 ---- aether/config.h | 10 - aether/global_ids.h | 1 - aether/tele/env/compilation_options.h | 2 - aether/uap/uap.cpp | 190 -------------- aether/uap/uap.h | 164 ------------ .../work_server_api/login_api.cpp | 1 - .../work_server_api/login_api.h | 1 - config/user_config_hydrogen.h | 2 - config/user_config_sodium.h | 2 - 18 files changed, 6 insertions(+), 818 deletions(-) delete mode 100644 aether/ae_actions/time_sync.cpp delete mode 100644 aether/ae_actions/time_sync.h delete mode 100644 aether/uap/uap.cpp delete mode 100644 aether/uap/uap.h diff --git a/aether/CMakeLists.txt b/aether/CMakeLists.txt index 22dcda42..13a303d1 100644 --- a/aether/CMakeLists.txt +++ b/aether/CMakeLists.txt @@ -70,9 +70,7 @@ list(APPEND aether_srcs "ae_actions/ping.cpp" "ae_actions/check_access_for_send_message.cpp" "ae_actions/telemetry.cpp" - "ae_actions/select_client.cpp" - "ae_actions/time_sync.cpp" - ) + "ae_actions/select_client.cpp") list(APPEND aether_srcs "registration/api/client_reg_api_safe.cpp" @@ -86,10 +84,6 @@ list(APPEND aether_srcs "registration/registration_crypto_provider.cpp" "registration/root_server_select_stream.cpp") -list(APPEND aether_srcs - "uap/uap.cpp" - ) - list(APPEND aether_srcs "adapters/adapter.cpp" "adapters/wifi_adapter.cpp" diff --git a/aether/ae_actions/time_sync.cpp b/aether/ae_actions/time_sync.cpp deleted file mode 100644 index 718f3a59..00000000 --- a/aether/ae_actions/time_sync.cpp +++ /dev/null @@ -1,246 +0,0 @@ -/* - * Copyright 2026 Aethernet Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "aether/ae_actions/time_sync.h" - -#if AE_TIME_SYNC_ENABLED - -# include - -# include "aether/client.h" -# include "aether/aether.h" -# include "aether/uap/uap.h" -# include "aether-miscpp/misc/override.h" -# include "aether/types/iterator.h" -# include "aether/executors/executors.h" -# include "aether/cloud_connections/cloud_server_connection.h" - -# include "aether/tele/tele.h" - -namespace ae { -namespace time_sync_internal { -static constexpr auto kRequestTimeout = 10s; -static constexpr auto kMaxTries = 5; - -auto TimeSyncRequest::EnsureConnected() { - return ex::create([&](auto& ctx) noexcept { - auto client_ptr = client_.Lock(); - if (!client_ptr) { - return ex::set_error(std::move(ctx.receiver), Failed{}); - } - - // check or subscribe for connection state to main server - client_ptr->cloud_connection().ForServers( - [&](CloudServerConnection* sc) { - assert((sc != nullptr) && "Server connection is null!"); - - auto* cc = sc->client_connection(); - assert((cc != nullptr) && "Client connection is null!"); - - // if already connected - if (cc->stream_info().link_state == LinkState::kLinked) { - return ex::set_value(std::move(ctx.receiver), Success{}); - } - - // wait till connected - link_state_sub_ = cc->stream_update_event().Subscribe([&, cc]() { - switch (cc->stream_info().link_state) { - case LinkState::kLinked: - return ex::set_value(std::move(ctx.receiver), Success{}); - break; - case LinkState::kLinkError: - return ex::set_error(std::move(ctx.receiver), Retry{}); - break; - default: - break; - } - }); - }); - }); -} - -auto TimeSyncRequest::SyncRequest() { - return ex::let_value([&](Success) noexcept { - return ex::create([&](auto& ctx) noexcept { - auto client_ptr = client_.Lock(); - if (!client_ptr) { - return ex::set_error(std::move(ctx.receiver), Failed{}); - } - - // send get_time_utc request to main server - // get_time_utc return server utc time point in microseconds - client_ptr->cloud_connection().ForServers( - [&](CloudServerConnection* sc) { - assert((sc != nullptr) && "Server connection is null!"); - - auto* cc = sc->client_connection(); - assert(((cc != nullptr) && - (cc->stream_info().link_state == LinkState::kLinked)) && - "Client connection is not linked!"); - - auto& write_action = - cc->LoginApiCall(SubApi{[&](auto& api) { - AE_TELED_DEBUG("Make time sync request"); - response_sub_ = api->get_time_utc().Subscribe( - [&, request_time{Now()}](auto const& p) { - if (!p) { - return ex::set_error(std::move(ctx.receiver), - Retry{}); - } - HandleResponse( - std::chrono::milliseconds{ - static_cast(p.value())}, - request_time, Now()); - // time synced - return ex::set_value(std::move(ctx.receiver), - Success{}); - }); - }}); - write_action_sub_ = - write_action.status_event().Subscribe([&](auto status) { - if (status == WriteAction::Status::kFail) { - AE_TELED_ERROR("Time sync write error, retry"); - return ex::set_error(std::move(ctx.receiver), Retry{}); - } - }); - }); - - // use raw time to avoid sync jumps - request_time_ = Now(); - }); - }); -} - -TimeSyncRequest::TimeSyncRequest(AeContext const& ae_context, - Ptr const& client) - : ae_context_{ae_context}, client_{client} { - auto s = - ex::for_range(Range{1, kMaxTries}, - [&](auto) { - return EnsureConnected() | SyncRequest() | - ex::with_timeout(ae_context_, kRequestTimeout) | - ex::let_error(Override{ - [](Retry) noexcept { - AE_TELED_ERROR("Time sync retry"); - return ex::just(ex::for_continue); - }, - [](ex::TimeoutError) noexcept { - AE_TELED_ERROR("Time sync response timeout"); - return ex::just(ex::for_continue); - }, - [](auto&&...) noexcept { - AE_TELED_ERROR("Time sync failed"); - return ex::just_error(Failed{}); - }, - }); - }) | - ex::let_stopped([]() noexcept { return ex::just_error(Failed{}); }); - - waiter_.emplace( - ae_context_, std::move(s), - [&](std::optional> const& res) noexcept { - if (!res || !*res) { - AE_TELED_ERROR("Time sync failed"); - } - if (res && *res) { - AE_TELED_INFO("Time sync succeeded"); - } - Finish(); - }); -} - -void TimeSyncRequest::HandleResponse(std::chrono::milliseconds server_epoch, - TimePoint request_time, - TimePoint response_time) { - auto server_time = TimePoint{server_epoch}; - auto round_trip = response_time - request_time; - AE_TELED_DEBUG( - "Time sync roundtrip {:%S} request_time {:%Y-%m-%d %H:%M:%S}, " - "response_time {:%Y-%m-%d %H:%M:%S} server_time {:%Y-%m-%d %H:%M:%S}", - std::chrono::duration_cast(round_trip), request_time, - response_time, server_time); - - auto diff_time = server_time - request_time - round_trip / 2; - AE_TELED_INFO( - "Time sync diff_time is {} ms", - std::chrono::duration_cast(diff_time).count()); - // update diff time - SyncClock::SyncTimeDiff += - std::chrono::duration_cast(diff_time); - AE_TELED_DEBUG("Current time {:%Y-%m-%d %H:%M:%S}", SyncClock::now()); -} - -} // namespace time_sync_internal - -// set end of time - this means last_sync_time is not set -TimePoint TimeSyncAction::last_sync_time = TimePoint::max(); - -TimeSyncAction::TimeSyncAction(AeContext const& ae_context, - Ptr const& client, - Duration sync_interval) - : ae_context_{ae_context}, client_{client}, sync_interval_{sync_interval} { - AE_TELED_INFO("Time sync created"); - // the end of time! It means never synced before - if (last_sync_time == TimePoint::max()) { - MakeRequest(); - } else { - ScheduleNextSync(); - } -} - -void TimeSyncAction::MakeRequest() { - auto uap = ae_context_.aether().uap.Load(); - if (!uap) { - return; - } - - // send request only if SendReceive Uap interval enabled - if (auto timer = uap->timer(); - timer.has_value() && - timer->interval().interval.type != IntervalType::kSendReceive) { - ScheduleNextSync(); - return; - } - - auto client_ptr = client_.Lock(); - if (!client_ptr) { - return; - } - - assert((!time_sync_request_ || time_sync_request_->is_finished()) && - "Time sync request already in progress"); - time_sync_request_.emplace(ae_context_, client_ptr); - uap->RegisterStart(); - time_sync_request_->finished_event().Subscribe( - [uap]() { uap->RegisterEnd(); }); - - // use raw time to avoid sync jumps - last_sync_time = Now(); - ScheduleNextSync(); -} - -void TimeSyncAction::ScheduleNextSync() { - TimePoint next_time = - ((last_sync_time == TimePoint::max()) ? Now() : last_sync_time) + - sync_interval_; - - task_sub_ = ae_context_.scheduler().DelayedTask([this]() { MakeRequest(); }, - next_time); -} -} // namespace ae -#endif diff --git a/aether/ae_actions/time_sync.h b/aether/ae_actions/time_sync.h deleted file mode 100644 index 8556308e..00000000 --- a/aether/ae_actions/time_sync.h +++ /dev/null @@ -1,87 +0,0 @@ -/* - * Copyright 2026 Aethernet Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#ifndef AETHER_AE_ACTIONS_TIME_SYNC_H_ -#define AETHER_AE_ACTIONS_TIME_SYNC_H_ - -#include "aether/config.h" - -#if AE_TIME_SYNC_ENABLED - -# include -# include "aether/env.h" -# include "aether/clock.h" -# include "aether/ae_context.h" -# include "aether/ptr/ptr_view.h" -# include "aether/actions/action.h" -# include "aether/executors/executors.h" - -namespace ae { -class Client; - -namespace time_sync_internal { -struct Success {}; -struct Failed {}; -struct Retry {}; - -class TimeSyncRequest : public Action { - public: - TimeSyncRequest(AeContext const& ae_context, Ptr const& client); - - private: - auto EnsureConnected(); - auto SyncRequest(); - static void HandleResponse(std::chrono::milliseconds server_epoch, - TimePoint request_time, TimePoint response_time); - - AeContext ae_context_; - PtrView client_; - TimePoint request_time_; - Subscription link_state_sub_; - Subscription response_sub_; - Subscription write_action_sub_; - std::optional< - ex::AnyWaiter> - waiter_; -}; -} // namespace time_sync_internal - -class TimeSyncAction { - enum class State : char { - kMakeRequest, - kWaitInterval, - kFailed, - }; - - public: - TimeSyncAction(AeContext const& ae_context, Ptr const& client, - Duration sync_interval); - - private: - void MakeRequest(); - void ScheduleNextSync(); - - AeContext ae_context_; - PtrView client_; - Duration sync_interval_; - std::optional time_sync_request_; - TaskSubscription task_sub_; - - static RTC_STORAGE_ATTR TimePoint last_sync_time; -}; -} // namespace ae -#endif -#endif // AETHER_AE_ACTIONS_TIME_SYNC_H_ diff --git a/aether/aether.cpp b/aether/aether.cpp index dfb9dd74..63bde5d6 100644 --- a/aether/aether.cpp +++ b/aether/aether.cpp @@ -19,8 +19,6 @@ #include #include "aether/obj/obj_ptr.h" -#include "aether/ae_actions/time_sync.h" - #include "aether/client.h" #include "aether/server.h" #include "aether/registration_cloud.h" @@ -54,7 +52,6 @@ Client::ptr Aether::CreateClient(ClientConfig const& config, std::string const& client_id) { auto client = FindClient(client_id); if (client.is_valid()) { - MakeTimeSyncAction(client); return client; } // create new client @@ -88,7 +85,6 @@ Client::ptr Aether::CreateClient(ClientConfig const& config, }); assert(res && "Failed to set client config"); - MakeTimeSyncAction(client); StoreClient(client); return client; } @@ -104,7 +100,6 @@ SelectClientAction& Aether::SelectClient([[maybe_unused]] Uid parent_uid, auto client = FindClient(client_id); if (client.is_valid()) { - MakeTimeSyncAction(client); return MakeSelectClient(client); } // register new client @@ -195,19 +190,4 @@ Registration& Aether::RegisterClient(std::string const& client_id, } #endif -void Aether::MakeTimeSyncAction([[maybe_unused]] Client::ptr const& client) { -#if AE_TIME_SYNC_ENABLED - if (time_sync_action_) { - return; - } - - client.WithLoaded([this](auto const& c) { - static constexpr auto kTimeSyncInterval = - std::chrono::seconds{AE_TIME_SYNC_INTERVAL_S}; - time_sync_action_ = - std::make_unique(*this, c, kTimeSyncInterval); - }); -#endif -} - } // namespace ae diff --git a/aether/aether.h b/aether/aether.h index 805031f4..43a090fb 100644 --- a/aether/aether.h +++ b/aether/aether.h @@ -28,8 +28,6 @@ #include "aether/ae_context.h" #include "aether/ae_actions/select_client.h" -#include "aether/uap/uap.h" - namespace ae { class Server; class Client; @@ -37,7 +35,6 @@ class Crypto; class IPoller; class DnsResolver; class Registration; -class TimeSyncAction; class AdapterRegistry; class ActionProcessor; class RegistrationCloud; @@ -59,22 +56,22 @@ class Aether : public Obj { ~Aether() override; AE_OBJECT_REFLECT(AE_MMBRS(client_prefab, registration_cloud, crypto, - clients_, servers_, tele_statistics, poller, - dns_resolver, adapter_registry, uap, - select_client_actions_)) + clients_, servers_, tele_statistics, poller, + dns_resolver, adapter_registry, + select_client_actions_)) template void Load(CurrentVersion, Dnv& dnv) { dnv(base_); dnv(client_prefab, registration_cloud, crypto, clients_, servers_, - tele_statistics, poller, dns_resolver, adapter_registry, uap); + tele_statistics, poller, dns_resolver, adapter_registry); } template void Save(CurrentVersion, Dnv& dnv) const { dnv(base_); dnv(client_prefab, registration_cloud, crypto, clients_, servers_, - tele_statistics, poller, dns_resolver, adapter_registry, uap); + tele_statistics, poller, dns_resolver, adapter_registry); } // AeContext protocol @@ -96,7 +93,6 @@ class Aether : public Obj { Obj::ptr dns_resolver; Obj::ptr adapter_registry; - Uap::ptr uap; Obj::ptr tele_statistics; @@ -120,16 +116,11 @@ class Aether : public Obj { std::map> registrations_; #endif - void MakeTimeSyncAction(ObjPtr const& client); - std::map clients_; std::map servers_; std::map> select_client_actions_; -#if AE_TIME_SYNC_ENABLED - std::unique_ptr time_sync_action_; -#endif }; } // namespace ae diff --git a/aether/aether_app.cpp b/aether/aether_app.cpp index 1f2299f3..cf33c7e0 100644 --- a/aether/aether_app.cpp +++ b/aether/aether_app.cpp @@ -231,17 +231,6 @@ static DnsResolver::ptr DnsResolverFactory(AetherAppContext const& context) { # endif } -static Uap::ptr UapFactory(AetherAppContext const& context) { - auto uap = context.aether()->uap; - if (uap.is_valid()) { - return uap; - } - return Uap::ptr::Create(CreateWith{context.domain()} - .with_id(GlobalId::kUap) - .with_flags(ObjFlags::kUnloadedByDefault), - context.aether(), std::initializer_list{}); -} - static Client::ptr ClientPrefabFactory(AetherAppContext const& context) { auto client_prefab = context.aether()->client_prefab; if (client_prefab.is_valid()) { @@ -301,10 +290,6 @@ void AetherAppContext::InitComponentContext() { dns_resolver_.Factory(::ae::DnsResolverFactory); } - if (!uap_) { - uap_.Factory(::ae::UapFactory); - } - if (!client_prefab_) { client_prefab_.Factory(::ae::ClientPrefabFactory); } @@ -321,7 +306,6 @@ RcPtr AetherApp::Construct(AetherAppContext context) { #if AE_DISTILLATION app->aether_->tele_statistics = context.tele_statistics_.Resolve(context); app->aether_->client_prefab = context.client_prefab_.Resolve(context); - app->aether_->uap = context.uap_.Resolve(context); app->aether_->adapter_registry = context.adapter_registry(); diff --git a/aether/aether_app.h b/aether/aether_app.h index e2acece3..de8cbf26 100644 --- a/aether/aether_app.h +++ b/aether/aether_app.h @@ -36,7 +36,6 @@ #include "aether/aether.h" #include "aether/crypto.h" #include "aether/client.h" -#include "aether/uap/uap.h" #include "aether/ae_context.h" #include "aether/poller/poller.h" #include "aether/dns/dns_resolve.h" @@ -90,8 +89,6 @@ class AetherAppContext { DnsResolver::ptr& dns_resolver() const { return dns_resolver_.Resolve(*this); } - Uap::ptr& uap() const { return uap_.Resolve(*this); } - #if AE_DISTILLATION template AetherAppContext&& AdaptersFactory(TFunc&& func) && { @@ -133,11 +130,6 @@ class AetherAppContext { return std::move(*this); } # endif - template - AetherAppContext&& UapFactory(TFunc&& func) && { - uap_.Factory(std::forward(func)); - return std::move(*this); - } #endif // AE_DISTILLATION private: @@ -152,7 +144,6 @@ class AetherAppContext { ComponentFactory crypto_; ComponentFactory poller_; ComponentFactory dns_resolver_; - ComponentFactory uap_; ComponentFactory client_prefab_; ComponentFactory tele_statistics_; diff --git a/aether/all.h b/aether/all.h index 1a9a4e1a..0f63ffe2 100644 --- a/aether/all.h +++ b/aether/all.h @@ -107,7 +107,6 @@ #include "aether/cloud.h" #include "aether/work_cloud.h" #include "aether/registration_cloud.h" -#include "aether/uap/uap.h" #include "aether/client_messages/p2p_message_stream.h" #include "aether/client_messages/p2p_safe_message_stream.h" diff --git a/aether/clock.h b/aether/clock.h index 5f8940ad..efcc4d52 100644 --- a/aether/clock.h +++ b/aether/clock.h @@ -20,45 +20,6 @@ #include #include -#include "aether/env.h" - -namespace ae::clock_internal { -using std::chrono::duration; -using std::chrono::duration_cast; -using std::chrono::system_clock; - -template -class SyncClock { - public: - static RTC_STORAGE_ATTR system_clock::duration SyncTimeDiff; - - using internal_clock = ChronoClock; - using rep = typename ChronoClock::rep; - using period = typename ChronoClock::period; - using duration = typename ChronoClock::duration; - - using time_point = std::chrono::time_point; - - static constexpr bool is_steady = ChronoClock::is_steady; - - /** - * \brief Get the current time with SyncTimeDiff - */ - static time_point now() { - return time_point{ChronoClock::now().time_since_epoch() + SyncTimeDiff}; - } - - static auto ToRawTime(time_point tp) { - return - typename ChronoClock::time_point{tp.time_since_epoch() - SyncTimeDiff}; - } -}; - -template -std::chrono::system_clock::duration SyncClock::SyncTimeDiff = - std::chrono::milliseconds{0}; -} // namespace ae::clock_internal - namespace ae { using std::chrono_literals::operator""h; using std::chrono_literals::operator""min; @@ -71,16 +32,10 @@ using std::chrono_literals::operator""ns; */ using Duration = std::chrono::duration; using SystemClock = std::chrono::system_clock; -using SyncClock = clock_internal::SyncClock; using TimePoint = typename SystemClock::time_point; -using SyncTimePoint = typename SyncClock::time_point; // current system clock time, without synchorinization inline auto Now() { return SystemClock::now(); } -// synchronised time -inline auto SyncTime() { return SyncClock::now(); } - -inline auto ToRawTime(SyncTimePoint tp) { return SyncClock::ToRawTime(tp); } } // namespace ae diff --git a/aether/config.h b/aether/config.h index 220927a1..17b31222 100644 --- a/aether/config.h +++ b/aether/config.h @@ -307,16 +307,6 @@ # define AE_CLOUD_REQUEST_TIMEOUT_MS 5000 #endif -// Time synchronization enabled -#ifndef AE_TIME_SYNC_ENABLED -# define AE_TIME_SYNC_ENABLED 1 -#endif - -// Time synchronization interval in seconds -#ifndef AE_TIME_SYNC_INTERVAL_S -# define AE_TIME_SYNC_INTERVAL_S 4 * 60 * 60 // every 4 hours -#endif - // Telemetry configuration // Compilation info // Environment info diff --git a/aether/global_ids.h b/aether/global_ids.h index 643477f3..28591217 100644 --- a/aether/global_ids.h +++ b/aether/global_ids.h @@ -31,7 +31,6 @@ struct GlobalId { static constexpr ObjId kRegistrationCloud{2}; static constexpr ObjId kTeleStatistics{3}; static constexpr ObjId kAdapterRegistry{4}; - static constexpr ObjId kUap{5}; static constexpr ObjId kEthernetAdapter = kGlobalIdAdaptersOffset + 1; static constexpr ObjId kLanAdapter = kGlobalIdAdaptersOffset + 2; static constexpr ObjId kWiFiAdapter = kGlobalIdAdaptersOffset + 3; diff --git a/aether/tele/env/compilation_options.h b/aether/tele/env/compilation_options.h index cb6e33b4..e7bbad0e 100644 --- a/aether/tele/env/compilation_options.h +++ b/aether/tele/env/compilation_options.h @@ -141,8 +141,6 @@ constexpr inline auto _compile_options_list = std::array{ _OPTION(AE_MODEM_CONNECTION_TIMEOUT_MS), _OPTION(AE_CLOUD_MAX_SERVER_CONNECTIONS), _OPTION(AE_CLOUD_SERVER_QUARANTINE_TIME_MS), - _OPTION(AE_TIME_SYNC_ENABLED), - _OPTION(AE_TIME_SYNC_INTERVAL_S), _OPTION(AE_TELE_ENABLED), _OPTION(AE_TELE_COMPILATION_INFO), _OPTION(AE_TELE_RUNTIME_INFO), diff --git a/aether/uap/uap.cpp b/aether/uap/uap.cpp deleted file mode 100644 index d0250cc9..00000000 --- a/aether/uap/uap.cpp +++ /dev/null @@ -1,190 +0,0 @@ -/* - * Copyright 2026 Aethernet Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "aether/uap/uap.h" - -#include -#include - -#include "aether/aether.h" - -#include "aether/tele/tele.h" - -namespace ae { -Duration Uap::IntervalState::remaining() const { - auto interval_end = until(); - auto current_time = Now(); - auto diff = interval_end - current_time; - return std::chrono::duration_cast(diff); -} - -TimePoint Uap::IntervalState::until() const { return end_time; } - -Uap::Timer::Timer(Uap::ptr uap) : uap_{std::move(uap)} {} - -Uap::IntervalState Uap::Timer::interval(Duration time_offset) const { - return uap_ - .WithLoaded( - [&](auto const& uap) { return uap->UpdateInterval(time_offset); }) - .value_or(Uap::IntervalState{}); -} - -Uap::Uap() { start_time_ = SyncTime(); } - -Uap::Uap(ObjProp prop, ObjPtr aether, - std::initializer_list const& interval_list) - : Obj{prop}, - aether_{std::move(aether)}, - intervals_{std::begin(interval_list), std::end(interval_list)} { - start_time_ = SyncTime(); - WindowWatcher(); -} - -Uap::~Uap() = default; - -void Uap::SleepReady() { - ready_to_sleep_ = true; - // if all registered actions is finished else wait /see RegisterAction - if (wait_actions_cnt_ == 0) { - AllActionsFinished(); - } -} - -Uap::SleepEvent::Subscriber Uap::sleep_event() { - return EventSubscriber{sleep_event_}; -} - -void Uap::SetIntervals(std::initializer_list const& interval_list) { - intervals_ = - std::vector{std::begin(interval_list), std::end(interval_list)}; - // update indices - current_interval_index_ = current_interval_index_ % intervals_.size(); - next_interval_index_ = (current_interval_index_ + 1) % intervals_.size(); -} - -std::optional Uap::timer() { - if (intervals_.empty()) { - return std::nullopt; - } - return Timer{Uap::ptr::MakeFromThis(this)}; -} - -void Uap::RegisterStart() { wait_actions_cnt_++; } -void Uap::RegisterEnd() { - assert(wait_actions_cnt_ > 0); - wait_actions_cnt_--; - if (wait_actions_cnt_ == 0) { - AllActionsFinished(); - } -} - -void Uap::Loaded() { WindowWatcher(); } - -void Uap::GoToSleep() { - if (intervals_.empty()) { - return; - } - sleep_event_.Emit(Timer{Uap::ptr::MakeFromThis(this)}); -} - -Uap::IntervalState Uap::UpdateInterval(Duration time_offset) { - assert(!intervals_.empty()); - - next_interval_index_ = (current_interval_index_ + 1) % intervals_.size(); - auto interval_duration = intervals_[current_interval_index_].duration; - auto current_time = SyncTime() + time_offset; - auto time_elapsed = current_time - start_time_; - AE_TELED_DEBUG( - "Update interval\n start_time {:%Y-%m-%d %H:%M:%S}\n current_time " - "{:%Y-%m-%d %H:%M:%S}\n interval_duration {:%H:%M:%S}\n time_elapsed " - "{:%H:%M:%S}", - start_time_, current_time, interval_duration, time_elapsed); - - if (time_elapsed > interval_duration) { - AE_TELED_WARNING( - "!More time elapsed than current interval {} > " - "{:%H:%M:%S}", - time_elapsed, interval_duration); - - std::size_t index = next_interval_index_; - auto interval_start = start_time_; - - // test for bad case - if time_elapsed is more than several uap durations - // find current interval index and new interval duration - auto uap_duration = std::accumulate( - std::begin(intervals_), std::end(intervals_), Duration{}, - [](auto v, auto const& i) { return v + i.duration; }); - // count how many full uap durations have elapsed and remove it from - // time_elapsed - auto uap_count = static_cast(time_elapsed / uap_duration); - time_elapsed -= uap_duration * uap_count; - interval_start += uap_duration * uap_count; - index = (index + uap_count * intervals_.size()) % intervals_.size(); - - // find the new interval - while (time_elapsed > interval_duration) { - // move all timers like it's new interval has started already - interval_start += interval_duration; - time_elapsed -= interval_duration; - index = (index + 1) % intervals_.size(); - interval_duration = intervals_[index].duration; - } - start_time_ = interval_start; - current_interval_index_ = index; - next_interval_index_ = (index + 1) % intervals_.size(); - } - AE_TELED_DEBUG( - "Current interval {}, next interval {}, start_time {:%Y-%m-%d %H:%M:%S} " - "current_time {:%Y-%m-%d %H:%M:%S}", - current_interval_index_, next_interval_index_, start_time_, current_time); - return IntervalState{ - .interval = intervals_[current_interval_index_], - .end_time = - ToRawTime(start_time_ + intervals_[current_interval_index_].duration), - }; -} - -void Uap::WindowWatcher() { - if (intervals_.empty()) { - return; - } - auto& current = intervals_[current_interval_index_]; - if (current.window == Duration::zero()) { - return; - } - - aether_.WithLoaded([this, w{current.window}](auto const& a) { - wait_actions_cnt_++; - AeContext{*a}.scheduler().DelayedTask( - [&]() { - assert(wait_actions_cnt_ > 0); - wait_actions_cnt_--; - if (wait_actions_cnt_ == 0) { - AllActionsFinished(); - } - }, - w); - }); -} - -void Uap::AllActionsFinished() { - AE_TELED_DEBUG("All registered actions finished"); - if (ready_to_sleep_) { - GoToSleep(); - } -} - -} // namespace ae diff --git a/aether/uap/uap.h b/aether/uap/uap.h deleted file mode 100644 index 6b6308fa..00000000 --- a/aether/uap/uap.h +++ /dev/null @@ -1,164 +0,0 @@ -/* - * Copyright 2026 Aethernet Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#ifndef AETHER_UAP_UAP_H_ -#define AETHER_UAP_UAP_H_ - -#include -#include - -#include "aether/clock.h" -#include "aether/obj/obj.h" -#include "aether/events/events.h" - -namespace ae { -class Aether; - -enum class IntervalType : char { - /** - * \brief Send only interval. - * All the network staff should be created only for sending data. - */ - kSendOnly, - /** - * \brief Send and receive interval. - * This activation window might be used for both sending and receiving data. - */ - kSendReceive, - /** - * \brief Receive only interval. - * All the network staff should be created only for receiving data at this - * activation window. - */ - kReceiveOnly, -}; - -/** - * \brief Basic interval type. - * This shows the duration between two activation windows. - * E.g. 10 seconds interval means the application should wake up, do its job, - * stay active at least for window duration, - * and go to sleep for the time remaining from this 10 seconds. - */ -struct Interval { - AE_REFLECT_MEMBERS(type, duration, window) - - IntervalType type; - Duration duration; - Duration window = Duration::zero(); //< by default the window size is 0 -}; - -class Uap final : public Obj { - AE_OBJECT(Uap, Obj, 0) - Uap(); - - struct IntervalState { - Interval interval; - TimePoint end_time; - - Duration remaining() const; - TimePoint until() const; - }; - - public: - class Timer { - public: - explicit Timer(Uap::ptr uap); - - /** - * \brief Get current interval offset. - * \param time_offset Offset from current time is needed if you want to get - * interval state in future. - */ - Uap::IntervalState interval(Duration time_offset = {}) const; - - Uap::ptr uap_; - }; - - using SleepEvent = Event; - - Uap(ObjProp prop, ObjPtr aether, - std::initializer_list const& interval_list); - ~Uap() override; - - AE_OBJECT_REFLECT(AE_MMBRS(aether_)) - - // save next interval and load current interval on its place - template - void Load(CurrentVersion, Dnv& dnv) { - dnv(base_, aether_, current_interval_index_, intervals_); - Loaded(); - } - template - void Save(CurrentVersion, Dnv& dnv) const { - dnv(base_, aether_, next_interval_index_, intervals_); - } - - /** - * \brief User notifies UAP - busyness logic is ready to sleep. - */ - void SleepReady(); - - /** - * \brief Uap notifies user - all systems are ready to sleep up to sleep_until - * time. - */ - SleepEvent::Subscriber sleep_event(); - - /** - * \brief Set the intervals in order they should be applied \see Interval. - */ - void SetIntervals(std::initializer_list const& interval_list); - - /** - * \brief Get timer. - * Timer allows to get current active interval. - */ - std::optional timer(); - - /** - * \brief Register action which should be finished before sleep event. - */ - void RegisterStart(); - void RegisterEnd(); - - private: - void Loaded(); - - void GoToSleep(); - /** - * \brief Get updated interval state \see Timer - */ - IntervalState UpdateInterval(Duration time_offset); - - // starts a special watcher to block GoToSleep on window duration - void WindowWatcher(); - // called when all registered actions is finished - void AllActionsFinished(); - - SleepEvent sleep_event_; - - ObjPtr aether_; - std::vector intervals_; - std::size_t current_interval_index_{}; - std::size_t next_interval_index_{}; - SyncTimePoint start_time_; - std::size_t wait_actions_cnt_{}; - bool ready_to_sleep_{false}; -}; -} // namespace ae - -#endif // AETHER_UAP_UAP_H_ diff --git a/aether/work_cloud_api/work_server_api/login_api.cpp b/aether/work_cloud_api/work_server_api/login_api.cpp index d76b28aa..6d6a248c 100644 --- a/aether/work_cloud_api/work_server_api/login_api.cpp +++ b/aether/work_cloud_api/work_server_api/login_api.cpp @@ -22,7 +22,6 @@ namespace ae { LoginApi::LoginApi(ProtocolContext& protocol_context, IEncryptProvider& encrypt_provider) : ApiClass{protocol_context}, - get_time_utc{protocol_context}, login_by_uid{protocol_context, LoginProc{*this}}, login_by_alias{protocol_context, LoginProc{*this}}, get_my_ip{protocol_context}, diff --git a/aether/work_cloud_api/work_server_api/login_api.h b/aether/work_cloud_api/work_server_api/login_api.h index a71e4899..a41bf102 100644 --- a/aether/work_cloud_api/work_server_api/login_api.h +++ b/aether/work_cloud_api/work_server_api/login_api.h @@ -44,7 +44,6 @@ class LoginApi : public ApiClass { explicit LoginApi(ProtocolContext& protocol_context, IEncryptProvider& encrypt_provider); - Method<3, ApiPromise()> get_time_utc; Method<4, void(Uid uid, SubApi sub_api), LoginProc> login_by_uid; Method<5, void(Uid alias, SubApi sub_api), LoginProc> diff --git a/config/user_config_hydrogen.h b/config/user_config_hydrogen.h index e20dacfa..981ec3b7 100644 --- a/config/user_config_hydrogen.h +++ b/config/user_config_hydrogen.h @@ -33,8 +33,6 @@ # define AE_SUPPORT_WIFIS 0 #endif -#define AE_TIME_SYNC_ENABLED 1 - // telemetry #define AE_TELE_ENABLED 1 #define AE_TELE_LOG_CONSOLE 1 diff --git a/config/user_config_sodium.h b/config/user_config_sodium.h index 1795a132..0b080110 100644 --- a/config/user_config_sodium.h +++ b/config/user_config_sodium.h @@ -33,8 +33,6 @@ # define AE_SUPPORT_WIFIS 0 #endif -#define AE_TIME_SYNC_ENABLED 1 - // telemetry #define AE_TELE_ENABLED 1 #define AE_TELE_LOG_CONSOLE 1 From a283337a65c20a0f500c007d19daa1be280c822e Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Tue, 7 Jul 2026 17:03:43 +0500 Subject: [PATCH 09/10] change default config for ping interval --- aether/config.h | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/aether/config.h b/aether/config.h index 17b31222..58ad3113 100644 --- a/aether/config.h +++ b/aether/config.h @@ -18,8 +18,8 @@ #define AETHER_CONFIG_H_ // IWYU pragma: begin_exports -#include #include +#include #include "aether/config_consts.h" #if defined USER_CONFIG @@ -264,7 +264,7 @@ // default value used for ping timeout, until statistics are available #ifndef AE_DEFAULT_RESPONSE_TIMEOUT_MS -# define AE_DEFAULT_RESPONSE_TIMEOUT_MS 10000 +# define AE_DEFAULT_RESPONSE_TIMEOUT_MS 5000 #endif // Is periodic ping messages enabled From 4aebcdd6f675a08715af16c2927c1f9c3fe642ab Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Tue, 7 Jul 2026 17:03:58 +0500 Subject: [PATCH 10/10] add connectivity policy conf and reset on start --- examples/cloud/cloud_test.cpp | 30 ++++++++++++++++++++++++++---- 1 file changed, 26 insertions(+), 4 deletions(-) diff --git a/examples/cloud/cloud_test.cpp b/examples/cloud/cloud_test.cpp index bf2b3ee9..9b1a7e19 100644 --- a/examples/cloud/cloud_test.cpp +++ b/examples/cloud/cloud_test.cpp @@ -30,10 +30,10 @@ #endif // IWYU pragma: begin_keeps +#include "aether_construct_esp_wifi.h" +#include "aether_construct_ethernet.h" #include "aether_construct_lora_module.h" #include "aether_construct_modem.h" -#include "aether_construct_ethernet.h" -#include "aether_construct_esp_wifi.h" // IWYU pragma: end_keeps namespace ae::cloud_test { @@ -91,6 +91,29 @@ int AetherCloudExample() { // clients must be selected assert(client_a && client_b); + // setup connectivity timings + using namespace std::chrono_literals; + client_a->connectivity_policy()->ResetRxTimings(); + client_b->connectivity_policy()->ResetRxTimings(); + + client_a->connectivity_policy() + ->ConfigureRxTimings(ae::RequestPolicy::All{}) + .ForPriority<0>(ae::RxTimingConf::Every(5s)) +#if AE_CLOUD_MAX_SERVER_CONNECTIONS >= 3 + .ForPriority<1>(ae::RxTimingConf::Every(10s)) + .ForPriority<2>(ae::RxTimingConf::Every(20s)) +#endif + ; + + client_b->connectivity_policy() + ->ConfigureRxTimings(ae::RequestPolicy::All{}) + .ForPriority<0>(ae::RxTimingConf::Every(5s)) +#if AE_CLOUD_MAX_SERVER_CONNECTIONS >= 3 + .ForPriority<1>(ae::RxTimingConf::Every(10s)) + .ForPriority<2>(ae::RxTimingConf::Every(20s)) +#endif + ; + // Make clients messages exchange. int received_count = 0; int confirmed_count = 0; @@ -137,8 +160,7 @@ int AetherCloudExample() { auto p2p_stream = std::make_shared( *aether_app, client_b.Load(), client_a->uid(), std::move(handle)); auto sender_stream = ae::make_unique( - *aether_app, ae::cloud_test::kSafeStreamConfig, - std::move(p2p_stream)); + *aether_app, ae::cloud_test::kSafeStreamConfig, std::move(p2p_stream)); sender_stream->out_data_event().Subscribe([&](auto const& data) { auto str_response =