From 931dde7af4e4cd478536514f9d3a9b49267c6da3 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Wed, 24 Jun 2026 15:46:58 +0500 Subject: [PATCH 1/6] remove client connection manager --- aether/CMakeLists.txt | 1 - aether/client.cpp | 16 ++---- aether/client.h | 3 -- aether/client_messages/p2p_message_stream.cpp | 21 +++----- aether/client_messages/p2p_message_stream.h | 9 ++-- .../p2p_message_stream_manager.cpp | 1 - .../p2p_message_stream_manager.h | 1 - .../cloud_connections_tele.h | 6 --- .../cloud_server_connections.cpp | 21 +++++--- .../cloud_server_connections.h | 18 +++++-- .../client_connection_manager.cpp | 44 ----------------- .../client_connection_manager.h | 49 ------------------- 12 files changed, 42 insertions(+), 148 deletions(-) delete mode 100644 aether/connection_manager/client_connection_manager.cpp delete mode 100644 aether/connection_manager/client_connection_manager.h diff --git a/aether/CMakeLists.txt b/aether/CMakeLists.txt index be442990..9d579e17 100644 --- a/aether/CMakeLists.txt +++ b/aether/CMakeLists.txt @@ -206,7 +206,6 @@ list(APPEND aether_srcs "server_connections/server_connection.cpp") list(APPEND aether_srcs - "connection_manager/client_connection_manager.cpp" "connection_manager/get_cloud_aether.cpp" "connection_manager/client_cloud_manager.cpp" "connection_manager/server_connection_manager.cpp") diff --git a/aether/client.cpp b/aether/client.cpp index 64bf3487..acae8640 100644 --- a/aether/client.cpp +++ b/aether/client.cpp @@ -40,7 +40,8 @@ Uid const& Client::ephemeral_uid() const { return ephemeral_uid_; } ServerKeys* Client::server_state(ServerId server_id) { auto ss_it = server_keys_.find(server_id); if (ss_it == server_keys_.end()) { - auto [it, _] = server_keys_.emplace(server_id, ServerKeys{server_id, master_key_}); + auto [it, _] = + server_keys_.emplace(server_id, ServerKeys{server_id, master_key_}); ss_it = it; } return &ss_it->second; @@ -62,20 +63,11 @@ ServerConnectionManager& Client::server_connection_manager() { return *server_connection_manager_; } -ClientConnectionManager& Client::connection_manager() { - if (!client_connection_manager_) { - auto aether = Aether::ptr{aether_}; - client_connection_manager_ = std::make_unique( - cloud_.Load(), - server_connection_manager().GetServerConnectionFactory()); - } - return *client_connection_manager_; -} - CloudServerConnections& Client::cloud_connection() { if (!cloud_connection_) { cloud_connection_ = std::make_unique( - *aether_.Load().as(), connection_manager(), + *aether_.Load().as(), cloud_.Load(), + server_connection_manager().GetServerConnectionFactory(), AE_CLOUD_MAX_SERVER_CONNECTIONS); #if AE_TELE_ENABLED diff --git a/aether/client.h b/aether/client.h index 836dc1d7..12f9f113 100644 --- a/aether/client.h +++ b/aether/client.h @@ -31,7 +31,6 @@ #include "aether/connection_manager/client_cloud_manager.h" #include "aether/client_messages/p2p_message_stream_manager.h" #include "aether/connection_manager/server_connection_manager.h" -#include "aether/connection_manager/client_connection_manager.h" namespace ae { class Aether; @@ -58,7 +57,6 @@ class Client : public Obj { Cloud::ptr const& cloud() const; ClientCloudManager::ptr const& cloud_manager() const; ServerConnectionManager& server_connection_manager(); - ClientConnectionManager& connection_manager(); CloudServerConnections& cloud_connection(); P2pMessageStreamManager& message_stream_manager(); @@ -86,7 +84,6 @@ class Client : public Obj { ClientCloudManager::ptr client_cloud_manager_; std::unique_ptr server_connection_manager_; - std::unique_ptr client_connection_manager_; std::unique_ptr cloud_connection_; std::unique_ptr message_stream_manager_; diff --git a/aether/client_messages/p2p_message_stream.cpp b/aether/client_messages/p2p_message_stream.cpp index 74b2755c..076353ee 100644 --- a/aether/client_messages/p2p_message_stream.cpp +++ b/aether/client_messages/p2p_message_stream.cpp @@ -254,11 +254,12 @@ void P2pStream::ConnectSend() { auto& get_client_cloud = client_ptr->cloud_manager()->GetCloud(destination_); get_client_cloud_sub_ = get_client_cloud.result_event().Subscribe( - [this](Result&& result) { + [this, client_ptr](Result&& result) { if (result) { auto cloud = std::move(result).value(); - dest_conn_manager_ = MakeConnectionManager(cloud.Load()); - dest_cloud_conn_ = MakeDestinationCloudConn(*dest_conn_manager_); + dest_cloud_conn_ = MakeDestinationCloudConn( + cloud.Load(), client_ptr->server_connection_manager() + .GetServerConnectionFactory()); // TODO: add config for request policy message_send_stream_ = std::make_unique( @@ -276,19 +277,11 @@ void P2pStream::ConnectSend() { }); } -std::unique_ptr P2pStream::MakeConnectionManager( - Ptr const& cloud) { - auto client_ptr = client_.Lock(); - assert(client_ptr); - return std::make_unique( - cloud, - client_ptr->server_connection_manager().GetServerConnectionFactory()); -} - std::unique_ptr P2pStream::MakeDestinationCloudConn( - ClientConnectionManager& connection_manager) { + Ptr const& cloud, + std::unique_ptr factory) { return std::make_unique( - ae_context_, connection_manager, AE_CLOUD_MAX_SERVER_CONNECTIONS); + ae_context_, cloud, std::move(factory), AE_CLOUD_MAX_SERVER_CONNECTIONS); } WriteAction* P2pStream::OnWrite(AeMessage&& message) { diff --git a/aether/client_messages/p2p_message_stream.h b/aether/client_messages/p2p_message_stream.h index 7d856bf1..449ec815 100644 --- a/aether/client_messages/p2p_message_stream.h +++ b/aether/client_messages/p2p_message_stream.h @@ -27,7 +27,6 @@ #include "aether/cloud_connections/cloud_server_connections.h" #include "aether/connection_manager/client_cloud_manager.h" -#include "aether/connection_manager/client_connection_manager.h" namespace ae { class Client; @@ -64,18 +63,16 @@ class P2pStream final : public ByteIStream { void ConnectReceive(); void ConnectSend(); - std::unique_ptr MakeConnectionManager( - Ptr const& cloud); std::unique_ptr MakeDestinationCloudConn( - ClientConnectionManager& connection_manager); + Ptr const& cloud, + std::unique_ptr factory); WriteAction* OnWrite(AeMessage&& message); AeContext ae_context_; PtrView client_; Uid destination_{}; - // connection manager to destination cloud - std::unique_ptr dest_conn_manager_; + // connection to destination cloud std::unique_ptr dest_cloud_conn_; BufferWrite buffer_write_; std::unique_ptr message_send_stream_; diff --git a/aether/client_messages/p2p_message_stream_manager.cpp b/aether/client_messages/p2p_message_stream_manager.cpp index a996dcb2..1ef78528 100644 --- a/aether/client_messages/p2p_message_stream_manager.cpp +++ b/aether/client_messages/p2p_message_stream_manager.cpp @@ -26,7 +26,6 @@ P2pMessageStreamManager::P2pMessageStreamManager(AeContext const& ae_context, Ptr const& client) : ae_context_{ae_context}, client_{client}, - connection_manager_{&client->connection_manager()}, cloud_connection_{&client->cloud_connection()}, on_message_received_sub_{CloudSubscription{ ClientListener{[this](ClientApiSafe& client_api, auto*) { diff --git a/aether/client_messages/p2p_message_stream_manager.h b/aether/client_messages/p2p_message_stream_manager.h index 64c623d8..d5204b50 100644 --- a/aether/client_messages/p2p_message_stream_manager.h +++ b/aether/client_messages/p2p_message_stream_manager.h @@ -50,7 +50,6 @@ class P2pMessageStreamManager { AeContext ae_context_; PtrView client_; - ClientConnectionManager* connection_manager_; CloudServerConnections* cloud_connection_; std::map> streams_; NewStreamEvent new_stream_event_; diff --git a/aether/cloud_connections/cloud_connections_tele.h b/aether/cloud_connections/cloud_connections_tele.h index 4ffc3c78..4bdf67f4 100644 --- a/aether/cloud_connections/cloud_connections_tele.h +++ b/aether/cloud_connections/cloud_connections_tele.h @@ -28,10 +28,4 @@ AE_TAG(CloudClientNewStream, kCloudClientConnection) AE_TELE_MODULE(kClientServerStream, 61, 113, 113); AE_TAG(ClientServerStreamCreate, kClientServerStream) -AE_TELE_MODULE(kClientConnectionManager, 62, 114, 116); -AE_TAG(ClientConnectionManagerSelfCloudConnection, kClientConnectionManager) -AE_TAG(ClientConnectionManagerUidCloudConnection, kClientConnectionManager) -AE_TAG(ClientConnectionManagerUnableCreateClientServerConnection, - kClientConnectionManager) - #endif // AETHER_CLOUD_CONNECTIONS_CLOUD_CONNECTIONS_TELE_H_ diff --git a/aether/cloud_connections/cloud_server_connections.cpp b/aether/cloud_connections/cloud_server_connections.cpp index 7734a2c5..6e702673 100644 --- a/aether/cloud_connections/cloud_server_connections.cpp +++ b/aether/cloud_connections/cloud_server_connections.cpp @@ -24,11 +24,14 @@ namespace ae { CloudServerConnections::CloudServerConnections( - AeContext const& ae_context, ClientConnectionManager& connection_manager, + AeContext const& ae_context, Ptr const& cloud, + std::unique_ptr connection_factory, std::size_t max_connections) : ae_context_{ae_context}, - connection_manager_{&connection_manager}, + cloud_{cloud}, + connection_factory_{std::move(connection_factory)}, max_connections_{max_connections} { + InitServerConnections(); InitServers(); } @@ -57,6 +60,13 @@ void CloudServerConnections::Restream() { } } +void CloudServerConnections::InitServerConnections() { + server_connections_.clear(); + for (auto& server : cloud_->servers()) { + server_connections_.emplace_back(server.Load(), *connection_factory_); + } +} + void CloudServerConnections::InitServers() { AE_TELED_DEBUG("Init servers"); auto server_candidates = ServerCandidates(); @@ -112,8 +122,7 @@ void CloudServerConnections::SubscribeToServerState( auto bad_server = [this, sc{&server_connection}]() { // TODO: add the policy how to change the server priority on failure // put server in quarantine and make it the least prioritized - auto new_priority = - sc->priority() + connection_manager_->server_connections().size(); + auto new_priority = sc->priority() + server_connections_.size(); sc->EndConnection(new_priority); QuarantineTimer(*sc); UnselectServer(*sc); @@ -183,8 +192,8 @@ void CloudServerConnections::QuarantineTimer( std::vector CloudServerConnections::ServerCandidates() { std::vector servers; - servers.reserve(connection_manager_->server_connections().size()); - for (auto& s : connection_manager_->server_connections()) { + servers.reserve(server_connections_.size()); + for (auto& s : server_connections_) { if (s.quarantine()) { continue; } diff --git a/aether/cloud_connections/cloud_server_connections.h b/aether/cloud_connections/cloud_server_connections.h index 13fb873c..2ea28c53 100644 --- a/aether/cloud_connections/cloud_server_connections.h +++ b/aether/cloud_connections/cloud_server_connections.h @@ -17,10 +17,14 @@ #define AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTIONS_H_ #include +#include +#include "aether/cloud.h" +#include "aether/ptr/ptr.h" #include "aether/ae_context.h" #include "aether/events/events.h" -#include "aether/connection_manager/client_connection_manager.h" +#include "aether/cloud_connections/cloud_server_connection.h" +#include "aether/server_connections/iserver_connection_factory.h" namespace ae { @@ -30,9 +34,10 @@ class CloudServerConnections { public: using ServersUpdate = Event; - CloudServerConnections(AeContext const& ae_context, - ClientConnectionManager& connection_manager, - std::size_t max_connections); + CloudServerConnections( + AeContext const& ae_context, Ptr const& cloud, + std::unique_ptr connection_factory, + std::size_t max_connections); /** * \brief The event then top list of the servers were updated. @@ -52,6 +57,7 @@ class CloudServerConnections { void Restream(); private: + void InitServerConnections(); void InitServers(); void SelectServers(std::vector const& servers); void SubscribeToServerState(CloudServerConnection& server_connection); @@ -63,8 +69,10 @@ class CloudServerConnections { std::vector ServerCandidates(); AeContext ae_context_; - ClientConnectionManager* connection_manager_; + Ptr cloud_; + std::unique_ptr connection_factory_; std::size_t max_connections_; + std::vector server_connections_; // selected list of servers sorted by the priority std::vector selected_servers_; diff --git a/aether/connection_manager/client_connection_manager.cpp b/aether/connection_manager/client_connection_manager.cpp deleted file mode 100644 index 7876aab4..00000000 --- a/aether/connection_manager/client_connection_manager.cpp +++ /dev/null @@ -1,44 +0,0 @@ -/* - * Copyright 2025 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/connection_manager/client_connection_manager.h" - -#include "aether/cloud.h" - -namespace ae { -ClientConnectionManager::ClientConnectionManager( - Ptr const& cloud, - std::unique_ptr&& connection_factory) - : cloud_{cloud}, connection_factory_{std::move(connection_factory)} { - InitServerConnections(); -} - -std::vector& -ClientConnectionManager::server_connections() { - return server_connections_; -} - -void ClientConnectionManager::InitServerConnections() { - auto cloud = cloud_.Lock(); - assert(cloud); - server_connections_.reserve(cloud->servers().size()); - for (auto& server : cloud->servers()) { - assert(server.is_valid()); - server_connections_.emplace_back(server.Load(), *connection_factory_); - } -} - -} // namespace ae diff --git a/aether/connection_manager/client_connection_manager.h b/aether/connection_manager/client_connection_manager.h deleted file mode 100644 index eaae7bd1..00000000 --- a/aether/connection_manager/client_connection_manager.h +++ /dev/null @@ -1,49 +0,0 @@ -/* - * Copyright 2025 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_CONNECTION_MANAGER_CLIENT_CONNECTION_MANAGER_H_ -#define AETHER_CONNECTION_MANAGER_CLIENT_CONNECTION_MANAGER_H_ - -#include "aether/ptr/ptr.h" -#include "aether/ptr/ptr_view.h" - -#include "aether/cloud_connections/cloud_server_connection.h" -#include "aether/server_connections/iserver_connection_factory.h" - -namespace ae { -class Cloud; - -/** - * \brief Manager of all connections to the client's cloud - */ -class ClientConnectionManager { - public: - ClientConnectionManager( - Ptr const& cloud, - std::unique_ptr&& connection_factory); - - std::vector& server_connections(); - - private: - void InitServerConnections(); - - PtrView cloud_; - std::unique_ptr connection_factory_; - std::vector server_connections_; -}; -} // namespace ae - -#endif // AETHER_CONNECTION_MANAGER_CLIENT_CONNECTION_MANAGER_H_ From 3d3bddad6e84988c2f98b5d085e88c43e1b940a1 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Wed, 24 Jun 2026 15:47:04 +0500 Subject: [PATCH 2/6] add new agents --- opencode.json | 26 +++++++++++++++++++++----- 1 file changed, 21 insertions(+), 5 deletions(-) diff --git a/opencode.json b/opencode.json index 819eeeef..0c1a0a60 100644 --- a/opencode.json +++ b/opencode.json @@ -34,10 +34,13 @@ "coder": { "mode": "subagent", "description": "Write c++ code", - "prompt": "You are a highly qualified c++ developer. Your task is to write c++ code that solves the given problem. You get instruction from the team lead/architect. You only implement the ideas in code.", + "prompt": "You are a highly qualified c++ developer. Your task is to write c++ code that solves the given problem. You get instruction from the team lead/architect. You only implement the ideas in code. To verify builds ask @builder_cpp.", "permission": { "edit": "allow", - "bash": "ask", + "bash": { + "cmake*": "deny", + "*": "ask" + }, "external_directory": "deny", "repo_clone": "deny" } @@ -50,15 +53,28 @@ "edit": "deny", "bash": { "git log*": "allow", - "git diff*": "allow" + "git diff*": "allow", + "*": "deny" } }, "temperature": 0.1 }, - "team-lead-architect": { + "architect": { + "mode": "all", + "description": "Review code and reasone architecture", + "prompt": "You a c++ architect. You don't write code, you don't build, you don't test. Your task to analaize and create architecture. You analyze existing code and search for architecture flaws and points of improvements. You analyze requirements and crete best architecture to implement them.", + "permission": { + "edit": "deny", + "bash": { + "git log*": "allow", + "git diff*": "allow" + } + } + }, + "team-lead": { "mode": "primary", "description": "The main agent to rule the others on the way to work on code.", - "prompt": "You are team-lead architect. You don't write code, you don't build, you don't test. You manage agent team and architect solutions. You have @coder - to write actual code by your detailed instructions; @builder_cpp to validate builds, analyze compiler errors; @tester to run tests and analyze test logs; @code-reviewer to work in pair with @coder and check if everything made as it's intended.", + "prompt": "You are team-lead. You don't write code, you don't build, you don't test. You manage agent team and architect solutions. You have @architect - to create a solution based on requirements and existent code; @coder - to write actual code by your detailed instructions; @builder_cpp to validate builds, analyze compiler errors; @tester to run tests and analyze test logs; @code-reviewer to work in pair with @coder and check if everything made as it's intended.", "permission": { "edit": "deny", "bash": "deny" From 76489bf787aea729f8781a409fb4063f2bea32d0 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Thu, 25 Jun 2026 14:28:43 +0500 Subject: [PATCH 3/6] update agents --- AGENTS.md | 4 +++- opencode.json | 18 +++++++++--------- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 8e8f164e..f0949518 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -93,7 +93,9 @@ To run tests, go into `` and run `ctest . --progress -j -E "((sodium)|(hydro)|(bcrypt)).*" --output-on-failure`. Or run specific test by name from `/tests/run/`. - To run smoke test, run `/aether-client-cpp-cloud`. + To run smoke test, run `/aether-client-cpp-cloud`. + Notice! Run `aether-client-cpp-cloud` generates `state` dir there object state is saved. + Remove this `state` dir before run to make clean run. Keep it to run with previous state. ## Operational Rules - Do not analyze logs until everything is working fine. diff --git a/opencode.json b/opencode.json index 0c1a0a60..2e813a3a 100644 --- a/opencode.json +++ b/opencode.json @@ -34,13 +34,11 @@ "coder": { "mode": "subagent", "description": "Write c++ code", - "prompt": "You are a highly qualified c++ developer. Your task is to write c++ code that solves the given problem. You get instruction from the team lead/architect. You only implement the ideas in code. To verify builds ask @builder_cpp.", + "prompt": "You are a highly qualified c++ developer. Your task is to write c++ code that solves the given problem. You get instruction from the team lead/architect. You only implement the ideas in code. If you need verify your code ask @builder_cpp to do it for you.", "permission": { "edit": "allow", - "bash": { - "cmake*": "deny", - "*": "ask" - }, + "grep": "allow", + "bash": "deny", "external_directory": "deny", "repo_clone": "deny" } @@ -50,11 +48,12 @@ "description": "Review code and validate if it solves the problem", "prompt": "You are trained as a code reviewer. Your task is to review the code written by the other agents and validate if it solves the problem. Notice we are working on a cross-platform project for desktop and IoT devices, so pay attention to platform-specific, performance, and security concerns. Never try to fix the issues yourself, just report.", "permission": { + "grep": "allow", "edit": "deny", "bash": { + "*": "deny", "git log*": "allow", "git diff*": "allow", - "*": "deny" } }, "temperature": 0.1 @@ -62,7 +61,7 @@ "architect": { "mode": "all", "description": "Review code and reasone architecture", - "prompt": "You a c++ architect. You don't write code, you don't build, you don't test. Your task to analaize and create architecture. You analyze existing code and search for architecture flaws and points of improvements. You analyze requirements and crete best architecture to implement them.", + "prompt": "You a c++ architect. You don't write code, you don't build, you don't test. Your task to analaize and create architecture. You analyze existing code and search for architecture flaws and points of improvements. You analyze requirements and crete best architecture to implement them. But you're never implement those solution only report them.", "permission": { "edit": "deny", "bash": { @@ -74,11 +73,12 @@ "team-lead": { "mode": "primary", "description": "The main agent to rule the others on the way to work on code.", - "prompt": "You are team-lead. You don't write code, you don't build, you don't test. You manage agent team and architect solutions. You have @architect - to create a solution based on requirements and existent code; @coder - to write actual code by your detailed instructions; @builder_cpp to validate builds, analyze compiler errors; @tester to run tests and analyze test logs; @code-reviewer to work in pair with @coder and check if everything made as it's intended.", + "prompt": "You are team-lead. You don't write code, you don't build, you don't test. You manage team of highly qualified agents and thats all. You have @architect - to create a solution based on requirements and existent code; @coder - to write actual code by your detailed instructions; @builder_cpp to validate builds, analyze compiler errors; @tester to run tests and analyze test logs; @code-reviewer to work in pair with @coder and check if everything made as it's intended. Print plan first and wait for user approve.", "permission": { "edit": "deny", "bash": "deny" - } + }, + "temperature": 0.1 } } } From 8458440e7a91ed9f2f08aedd63c6c6ba1ab5200f Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Thu, 25 Jun 2026 14:30:14 +0500 Subject: [PATCH 4/6] make cloud better cloud request --- .../check_access_for_send_message.cpp | 2 +- .../check_access_for_send_message.h | 2 +- aether/ae_actions/get_servers.cpp | 39 +- aether/ae_actions/get_servers.h | 4 +- aether/ae_actions/telemetry.cpp | 15 +- aether/ae_actions/telemetry.h | 4 +- aether/ae_actions/time_sync.cpp | 11 +- aether/client_messages/p2p_message_stream.cpp | 26 +- .../p2p_message_stream_manager.cpp | 4 +- .../p2p_message_stream_manager.h | 2 +- aether/cloud_connections/cloud_callbacks.h | 33 +- aether/cloud_connections/cloud_request.cpp | 332 +++++++++--------- aether/cloud_connections/cloud_request.h | 99 ++---- .../cloud_server_connections.cpp | 116 +++++- .../cloud_server_connections.h | 95 ++++- .../cloud_connections/cloud_subscription.cpp | 28 +- aether/cloud_connections/cloud_subscription.h | 22 +- aether/cloud_connections/cloud_visit.h | 79 ----- aether/config.h | 5 + .../client_cloud_manager.cpp | 4 +- .../connection_manager/client_cloud_manager.h | 2 +- .../connection_manager/get_cloud_aether.cpp | 11 +- aether/connection_manager/get_cloud_aether.h | 3 +- 23 files changed, 514 insertions(+), 424 deletions(-) delete mode 100644 aether/cloud_connections/cloud_visit.h diff --git a/aether/ae_actions/check_access_for_send_message.cpp b/aether/ae_actions/check_access_for_send_message.cpp index f5db2409..6fbf0aa3 100644 --- a/aether/ae_actions/check_access_for_send_message.cpp +++ b/aether/ae_actions/check_access_for_send_message.cpp @@ -26,7 +26,7 @@ CheckAccessForSendMessage::CheckAccessForSendMessage( : destination_{destination}, cloud_request_{ ae_context, - AuthApiRequest{[this](ApiContext& auth_api, auto*, + ApiRequestHandler{[this](ApiContext& auth_api, auto*, auto* request) { wait_check_sub_ = auth_api->check_access_for_send_message(destination_) diff --git a/aether/ae_actions/check_access_for_send_message.h b/aether/ae_actions/check_access_for_send_message.h index 04ff648a..433989ce 100644 --- a/aether/ae_actions/check_access_for_send_message.h +++ b/aether/ae_actions/check_access_for_send_message.h @@ -46,7 +46,7 @@ class CheckAccessForSendMessage final : public Action { void ErrorReceived(); Uid destination_; - CloudRequestAction cloud_request_; + CloudRequest cloud_request_; ResultEvent result_event_; Subscription wait_check_sub_; }; diff --git a/aether/ae_actions/get_servers.cpp b/aether/ae_actions/get_servers.cpp index 75a48387..df7549bf 100644 --- a/aether/ae_actions/get_servers.cpp +++ b/aether/ae_actions/get_servers.cpp @@ -28,27 +28,30 @@ GetServersAction::GetServersAction(AeContext const& ae_context, : server_ids_{std::move(server_ids)}, cloud_request_{ ae_context, - AuthApiCaller{[this](ApiContext& auth_api, auto*) { - AE_TELED_DEBUG("Resolve servers {}", server_ids_); - auth_api->resolver_servers(server_ids_); - }}, - ClientResponseListener{[this](ClientApiSafe& client_api, auto*, + ApiCallWithListener{ + ApiCall{[this](ApiContext& auth_api, auto*) { + AE_TELED_DEBUG("Resolve servers {}", server_ids_); + auth_api->resolver_servers(server_ids_); + }}, + ResponseSubscriber{[this](ClientApiSafe& client_api, auto*, auto* request) { - return client_api.send_server_descriptor_event().Subscribe( - [this, request](auto const& sd) { GetResponse(sd, request); }); - }}, + return client_api.send_server_descriptor_event().Subscribe( + [this, request](auto const& sd) { + GetResponse(sd, request); + }); + }}}, cloud_connection, request_policy, } { - request_subs_ += cloud_request_.success_event().Subscribe([this]() { - AE_TELED_INFO("GetServersAction succeeded"); - result_event_.Emit( - Ok const&>{server_descriptors_}); - Finish(); - }); - request_subs_ += cloud_request_.failure_event().Subscribe([this]() { - AE_TELED_ERROR("GetServersAction failed"); - result_event_.Emit(Error{1}); + request_subs_ += cloud_request_.result_event().Subscribe([this](bool success) { + if (success) { + AE_TELED_INFO("GetServersAction succeeded"); + result_event_.Emit( + Ok const&>{server_descriptors_}); + } else { + AE_TELED_ERROR("GetServersAction failed"); + result_event_.Emit(Error{1}); + } Finish(); }); } @@ -58,7 +61,7 @@ GetServersAction::ResultEvent::Subscriber GetServersAction::result_event() { } void GetServersAction::GetResponse(ServerDescriptor const& server_descriptor, - CloudRequestAction* request) { + CloudRequest* request) { // If got not requested server id ignore it. if (auto it = std::find(std::begin(server_ids_), std::end(server_ids_), server_descriptor.server_id); diff --git a/aether/ae_actions/get_servers.h b/aether/ae_actions/get_servers.h index b78aef5a..ef7d34ff 100644 --- a/aether/ae_actions/get_servers.h +++ b/aether/ae_actions/get_servers.h @@ -43,13 +43,13 @@ class GetServersAction : public Action { private: void GetResponse(ServerDescriptor const& server_descriptor, - CloudRequestAction* request); + CloudRequest* request); std::vector server_ids_; TimePoint timeout_point_; ResultEvent result_event_; - CloudRequestAction cloud_request_; + CloudRequest cloud_request_; MultiSubscription request_subs_; std::vector server_descriptors_; diff --git a/aether/ae_actions/telemetry.cpp b/aether/ae_actions/telemetry.cpp index 0937d956..3c7d0d35 100644 --- a/aether/ae_actions/telemetry.cpp +++ b/aether/ae_actions/telemetry.cpp @@ -25,7 +25,6 @@ # include "aether/mstream.h" # include "aether/mstream_buffers.h" -# include "aether/cloud_connections/cloud_request.h" # include "aether/ae_actions/ae_actions_tele.h" @@ -34,14 +33,12 @@ Telemetry::Telemetry(AeContext const& ae_context, CloudServerConnections& cloud_connection) : ae_context_{ae_context}, cloud_connection_{&cloud_connection}, - call_request_{ae_context_, *cloud_connection_}, telemetry_request_sub_{ - ClientListener{[&](ClientApiSafe& api, auto* sever_connect) { + ApiEventSubscriber{[&](ClientApiSafe& api, auto* sever_connect) { return api.request_telemetry_event().Subscribe( [&]() { OnRequestTelemetry(sever_connect->priority()); }); }}, - *cloud_connection_, - RequestPolicy::Replica{cloud_connection_->max_connections()}} { + *cloud_connection_, RequestPolicy::All{}} { AE_TELE_INFO(TelemetryCreated); } @@ -51,13 +48,13 @@ void Telemetry::SendTelemetry() { auto server_num = request_for_server_.value_or(0); request_for_server_.reset(); - if (server_num >= cloud_connection_->servers().size()) { + if (server_num >= cloud_connection_->selected_servers().size()) { AE_TELED_ERROR("Requested server number is out of range"); return; } ClientServerConnection* con = - cloud_connection_->servers().at(server_num)->client_connection(); + cloud_connection_->selected_servers().at(server_num)->client_connection(); assert((con != nullptr) && "ClientServerConnection is null"); auto telemetry = CollectTelemetry(con->stream_info()); @@ -66,8 +63,8 @@ void Telemetry::SendTelemetry() { return; } - call_request_.CallApi( - AuthApiCaller{[&](ApiContext& auth_api, auto*) { + cloud_connection_->CallApi( + ApiCall{[&](ApiContext& auth_api, auto*) { auth_api->send_telemetry(std::move(*telemetry)); }}, RequestPolicy::Priority{server_num}); diff --git a/aether/ae_actions/telemetry.h b/aether/ae_actions/telemetry.h index 4aba55d9..fb3d1d88 100644 --- a/aether/ae_actions/telemetry.h +++ b/aether/ae_actions/telemetry.h @@ -25,7 +25,6 @@ # include "aether/ae_context.h" # include "aether/stream_api/istream.h" -# include "aether/cloud_connections/cloud_request.h" # include "aether/cloud_connections/cloud_subscription.h" # include "aether/cloud_connections/cloud_server_connections.h" @@ -49,9 +48,8 @@ class Telemetry { AeContext ae_context_; CloudServerConnections* cloud_connection_; - CloudRequest call_request_; - CloudSubscription telemetry_request_sub_; + CloudEventListener telemetry_request_sub_; std::optional request_for_server_; }; } // namespace ae diff --git a/aether/ae_actions/time_sync.cpp b/aether/ae_actions/time_sync.cpp index 2f9827c7..718f3a59 100644 --- a/aether/ae_actions/time_sync.cpp +++ b/aether/ae_actions/time_sync.cpp @@ -26,7 +26,6 @@ # include "aether-miscpp/misc/override.h" # include "aether/types/iterator.h" # include "aether/executors/executors.h" -# include "aether/cloud_connections/cloud_visit.h" # include "aether/cloud_connections/cloud_server_connection.h" # include "aether/tele/tele.h" @@ -45,7 +44,7 @@ auto TimeSyncRequest::EnsureConnected() { } // check or subscribe for connection state to main server - CloudVisit::Visit( + client_ptr->cloud_connection().ForServers( [&](CloudServerConnection* sc) { assert((sc != nullptr) && "Server connection is null!"); @@ -70,8 +69,7 @@ auto TimeSyncRequest::EnsureConnected() { break; } }); - }, - client_ptr->cloud_connection(), RequestPolicy::MainServer{}); + }); }); } @@ -86,7 +84,7 @@ auto TimeSyncRequest::SyncRequest() { // send get_time_utc request to main server // get_time_utc return server utc time point in microseconds - CloudVisit::Visit( + client_ptr->cloud_connection().ForServers( [&](CloudServerConnection* sc) { assert((sc != nullptr) && "Server connection is null!"); @@ -120,8 +118,7 @@ auto TimeSyncRequest::SyncRequest() { return ex::set_error(std::move(ctx.receiver), Retry{}); } }); - }, - client_ptr->cloud_connection(), RequestPolicy::MainServer{}); + }); // use raw time to avoid sync jumps request_time_ = Now(); diff --git a/aether/client_messages/p2p_message_stream.cpp b/aether/client_messages/p2p_message_stream.cpp index 076353ee..4e941085 100644 --- a/aether/client_messages/p2p_message_stream.cpp +++ b/aether/client_messages/p2p_message_stream.cpp @@ -23,7 +23,6 @@ #include "aether/cloud.h" #include "aether/client.h" -#include "aether/cloud_connections/cloud_visit.h" #include "aether/cloud_connections/cloud_request.h" #include "aether/cloud_connections/cloud_subscription.h" @@ -33,11 +32,9 @@ namespace ae { namespace p2p_stream_internal { class MessageSendStream final : public IStream { public: - explicit MessageSendStream(AeContext const& ae_context, - CloudServerConnections& cloud_connection, + explicit MessageSendStream(CloudServerConnections& cloud_connection, RequestPolicy::Variant request_policy) - : cloud_request_{ae_context, cloud_connection}, - cloud_connection_{&cloud_connection}, + : cloud_connection_{&cloud_connection}, request_policy_{request_policy}, servers_update_sub_{cloud_connection_->servers_update_event().Subscribe( MethodPtr<&MessageSendStream::UpdateServers>{this})} { @@ -45,8 +42,8 @@ class MessageSendStream final : public IStream { } WriteAction& Write(AeMessage&& message) override { - return cloud_request_.CallApi( - AuthApiCaller{[&message](ApiContext& auth_api, auto*) { + return cloud_connection_->CallApi( + ApiCall{[&message](ApiContext& auth_api, auto*) { auth_api->send_message(std::move(message)); }}, request_policy_); @@ -61,26 +58,26 @@ class MessageSendStream final : public IStream { private: void UpdateServers() { - CloudVisit::Visit( + cloud_connection_->ForServers( [this](auto* sc) { if (auto* con = sc->client_connection(); con != nullptr) { streams_update_sub_ += con->stream_update_event().Subscribe( MethodPtr<&MessageSendStream::UpdateStream>{this}); } }, - *cloud_connection_, request_policy_); + request_policy_); UpdateStream(); } void UpdateStream() { std::vector infos; - CloudVisit::Visit( + cloud_connection_->ForServers( [&](auto* sc) { if (auto* con = sc->client_connection(); con != nullptr) { infos.emplace_back(con->stream_info()); } }, - *cloud_connection_, request_policy_); + request_policy_); if (infos.empty()) { stream_info_ = {}; @@ -137,7 +134,6 @@ class MessageSendStream final : public IStream { stream_update_event_.Emit(); } - CloudRequest cloud_request_; CloudServerConnections* cloud_connection_; RequestPolicy::Variant request_policy_; @@ -157,7 +153,7 @@ class ReadMessageGate { RequestPolicy::Variant request_policy) : uid_{uid}, message_event_sub_{ - ClientListener{[this](ClientApiSafe& client_api, auto*) { + ApiEventSubscriber{[this](ClientApiSafe& client_api, auto*) { return client_api.send_message_event().Subscribe( [this](auto const& message) { if (message.uid == uid_) { @@ -173,7 +169,7 @@ class ReadMessageGate { private: Uid uid_; - CloudSubscription message_event_sub_; + CloudEventListener message_event_sub_; OutDataEvent out_data_event_; }; @@ -263,7 +259,7 @@ void P2pStream::ConnectSend() { // TODO: add config for request policy message_send_stream_ = std::make_unique( - ae_context_, *dest_cloud_conn_, RequestPolicy::MainServer{}); + *dest_cloud_conn_, RequestPolicy::MainServer{}); message_send_stream_->stream_update_event().Subscribe( stream_update_event_); AE_TELED_DEBUG("Send connected"); diff --git a/aether/client_messages/p2p_message_stream_manager.cpp b/aether/client_messages/p2p_message_stream_manager.cpp index 1ef78528..ef55e898 100644 --- a/aether/client_messages/p2p_message_stream_manager.cpp +++ b/aether/client_messages/p2p_message_stream_manager.cpp @@ -27,8 +27,8 @@ P2pMessageStreamManager::P2pMessageStreamManager(AeContext const& ae_context, : ae_context_{ae_context}, client_{client}, cloud_connection_{&client->cloud_connection()}, - on_message_received_sub_{CloudSubscription{ - ClientListener{[this](ClientApiSafe& client_api, auto*) { + on_message_received_sub_{CloudEventListener{ + ApiEventSubscriber{[this](ClientApiSafe& client_api, auto*) { return client_api.send_message_event().Subscribe( MethodPtr<&P2pMessageStreamManager::NewMessageReceived>{this}); }}, diff --git a/aether/client_messages/p2p_message_stream_manager.h b/aether/client_messages/p2p_message_stream_manager.h index d5204b50..46e97d41 100644 --- a/aether/client_messages/p2p_message_stream_manager.h +++ b/aether/client_messages/p2p_message_stream_manager.h @@ -53,7 +53,7 @@ class P2pMessageStreamManager { CloudServerConnections* cloud_connection_; std::map> streams_; NewStreamEvent new_stream_event_; - CloudSubscription on_message_received_sub_; + CloudEventListener on_message_received_sub_; MultiSubscription message_stream_update_subs_; }; } // namespace ae diff --git a/aether/cloud_connections/cloud_callbacks.h b/aether/cloud_connections/cloud_callbacks.h index 962734fc..8e830869 100644 --- a/aether/cloud_connections/cloud_callbacks.h +++ b/aether/cloud_connections/cloud_callbacks.h @@ -27,29 +27,36 @@ namespace ae { class CloudServerConnection; class CloudServerConnections; -class CloudRequestAction; +class CloudRequest; // subscribe to client's api events -struct ClientListener : SmallFunction {}; +struct ApiEventSubscriber : SmallFunction {}; // call authorized api -struct AuthApiCaller +struct ApiCall : SmallFunction& auth_api, CloudServerConnection* server_connection)> {}; -// make request to authorized api -struct AuthApiRequest - : SmallFunction& auth_api, - CloudServerConnection* server_connection, - CloudRequestAction* request)> {}; // listen to client's api response -struct ClientResponseListener +struct ResponseSubscriber : SmallFunction {}; + CloudRequest* request)> {}; + +// ApiCall combined with its ResponseSubscriber +struct ApiCallWithListener { + ApiCall call; + ResponseSubscriber listener; +}; + +// make request to authorized api (handles its own response) +struct ApiRequestHandler + : SmallFunction& auth_api, + CloudServerConnection* server_connection, + CloudRequest* request)> {}; } // namespace ae -#endif // AETHER_CLOUD_CONNECTIONS_CLOUD_CALLBACKS_H_ +#endif // AETHER_CLOUD_CONNECTIONS_CLOUD_CALLBACKS_H_ \ No newline at end of file diff --git a/aether/cloud_connections/cloud_request.cpp b/aether/cloud_connections/cloud_request.cpp index 4f658339..7ed6e4d4 100644 --- a/aether/cloud_connections/cloud_request.cpp +++ b/aether/cloud_connections/cloud_request.cpp @@ -16,223 +16,239 @@ #include "aether/cloud_connections/cloud_request.h" -#include #include -#include "aether/server.h" #include "aether/aether.h" -#include "aether-miscpp/reflect/reflect.h" +#include "aether/server.h" +#include "aether-miscpp/misc/override.h" #include "aether/write_action/write_action.h" -#include "aether/cloud_connections/cloud_visit.h" #include "aether/cloud_connections/cloud_connections_tele.h" namespace ae { -CloudRequest::ReplicaWA::ReplicaWA(std::vector&& swas) noexcept - : swas_{std::move(swas)} { - assert(!swas_.empty()); - for (auto* action : swas_) { - // get the state from replicas by OR - // any success or all ether failed - subs_ += - action->status_event().Subscribe([this](WriteAction::Status status) { - switch (status) { - case WriteAction::Status::kSuccess: - SetStatus(status); - return; - case WriteAction::Status::kFail: - failed_actions_++; - break; - case WriteAction::Status::kStop: - stopped_actions_++; - break; - } - if ((failed_actions_ + stopped_actions_) >= swas_.size()) { - SetStatus(status); - } - }); - } -} - -void CloudRequest::ReplicaWA::Stop() noexcept { - for (auto* action : swas_) { - action->Stop(); - } -} CloudRequest::CloudRequest(AeContext const& ae_context, - CloudServerConnections& connection) - : ae_context_{ae_context}, connection_{&connection} {} - -WriteAction& CloudRequest::CallApi(AuthApiCaller const& api_caller, - RequestPolicy::Variant policy) { - std::vector swas; - CloudVisit::Visit( - [&](CloudServerConnection* sc) { - auto* conn = sc->client_connection(); - assert((conn != nullptr) && "Client connection is null"); - swas.emplace_back(&conn->AuthorizedApiCall( - SubApi{[&](auto& api) { api_caller(api, sc); }})); - }, - *connection_, policy); - - if (swas.empty()) { - return EmptyWriteAction(); - } - if (swas.size() == 1) { - return *swas.front(); - } - return ReplicaWriteAction(std::move(swas)); -} - -WriteAction& CloudRequest::EmptyWriteAction() { - if (!empty_wa_ || empty_wa_->is_finished()) { - empty_wa_.emplace(ae_context_); - } - return *empty_wa_; -} - -WriteAction& CloudRequest::ReplicaWriteAction( - std::vector&& swas) { - // TODO: replace vector to something static - return replica_was_.emplace_back(std::move(swas)); -} - -CloudRequestAction::CloudRequestAction( - AeContext const& ae_context, AuthApiCaller&& api_caller, - ClientResponseListener&& listener, - CloudServerConnections& cloud_server_connections, - RequestPolicy::Variant policy) + ApiCallWithListener&& api_call, + CloudServerConnections& cloud_server_connections, + RequestPolicy::Variant policy, + std::size_t max_retries, Duration request_timeout) : ae_context_{ae_context}, - request_{std::move(api_caller)}, - listener_{std::move(listener)}, - cloud_sc_{&cloud_server_connections}, + request_{std::move(api_call)}, + cloud_scs_{&cloud_server_connections}, policy_{policy}, - server_changed_sub_{cloud_sc_->servers_update_event().Subscribe( - MethodPtr<&CloudRequestAction::ServersUpdated>{this})} { + max_retries_{max_retries}, + request_timeout_{request_timeout}, + server_changed_sub_{cloud_scs_->servers_update_event().Subscribe( + MethodPtr<&CloudRequest::ServersUpdated>{this})} { + PrefillServerRequests(); EnqueueMakeRequest(); } -CloudRequestAction::CloudRequestAction( - AeContext const& ae_context, AuthApiRequest&& api_request, - CloudServerConnections& cloud_server_connections, - RequestPolicy::Variant policy) +CloudRequest::CloudRequest(AeContext const& ae_context, + ApiRequestHandler&& api_request, + CloudServerConnections& cloud_server_connections, + RequestPolicy::Variant policy, + std::size_t max_retries, Duration request_timeout) : ae_context_{ae_context}, request_{std::move(api_request)}, - cloud_sc_{&cloud_server_connections}, + cloud_scs_{&cloud_server_connections}, policy_{policy}, - server_changed_sub_{cloud_sc_->servers_update_event().Subscribe( - MethodPtr<&CloudRequestAction::ServersUpdated>{this})} { + max_retries_{max_retries}, + request_timeout_{request_timeout}, + server_changed_sub_{cloud_scs_->servers_update_event().Subscribe( + MethodPtr<&CloudRequest::ServersUpdated>{this})} { + PrefillServerRequests(); EnqueueMakeRequest(); } -void CloudRequestAction::Succeeded() { +void CloudRequest::Succeeded() { Finish(); - success_event_.Emit(); + result_event_.Emit(true); } -void CloudRequestAction::Failed() { +void CloudRequest::Failed() { Finish(); - failure_event_.Emit(); + result_event_.Emit(false); } -CloudRequestAction::SuccessEvent::Subscriber -CloudRequestAction::success_event() { - return EventSubscriber{success_event_}; +CloudRequest::ResultEvent::Subscriber CloudRequest::result_event() { + return EventSubscriber{result_event_}; } -CloudRequestAction::FailureEvent::Subscriber -CloudRequestAction::failure_event() { - return EventSubscriber{failure_event_}; +void CloudRequest::PrefillServerRequests() { + for (auto* sc : cloud_scs_->servers()) { + server_requests_.emplace(sc, ServerRequest{}); + } } -void CloudRequestAction::MakeRequest() { - CloudVisit::Visit( +void CloudRequest::MakeRequest() { + cloud_scs_->ForServers( [&](auto& sc) { - auto [it, ok] = server_requests_.emplace(sc, ServerRequest{}); - if (!ok) { - return; + auto it = server_requests_.find(sc); + ServerRequest* sr; + if (it == server_requests_.end()) { + // New server added to cloud after construction + auto [new_it, ok] = server_requests_.emplace(sc, ServerRequest{}); + sr = &new_it->second; + } else { + sr = &it->second; + if (sr->exhausted) { + return; + } } - MakeServerRequest(it->first, it->second); + MakeServerRequest(sc, *sr); }, - *cloud_sc_, policy_); + policy_); + + // Check if all server requests are exhausted + bool all_exhausted = !server_requests_.empty(); + for (auto const& [sc, sr] : server_requests_) { + if (!sr.exhausted) { + all_exhausted = false; + break; + } + } + if (all_exhausted) { + AE_TELED_ERROR("All server requests exhausted, failing"); + Failed(); + } } -void CloudRequestAction::MakeServerRequest(CloudServerConnection* sc, - ServerRequest& sr) { +void CloudRequest::MakeServerRequest(CloudServerConnection* sc, + ServerRequest& sr) { + AE_TELED_DEBUG("Make request to server {}", sc->server()->server_id); + + // Clear previous subscriptions and timeout + sr.state_subs.Reset(); + sr.timeout_sub.Reset(); + auto* conn = sc->client_connection(); assert((conn != nullptr) && "Client connection is null"); - AE_TELED_DEBUG("Make request to server {}", sc->server()->server_id); - // make request depends on saved request kind - auto& swa = std::visit( // ~['_']~ - reflect::OverrideFunc{ - // Plain AuthApiCaller with ClientResponseListener - [&](AuthApiCaller& api_caller) -> decltype(auto) { - return conn->AuthorizedApiCall(SubApi{ - [&](ApiContext& api) { api_caller(api, sc); }}); - }, - // AuthApiRequest - [&](AuthApiRequest& api_request) -> decltype(auto) { - return conn->AuthorizedApiCall( - SubApi{[&](ApiContext& api) { - api_request(api, sc, this); - }}); - }, - }, - request_); + auto& swa = + std::visit(Override{ + // ApiCallWithListener + [&](ApiCallWithListener& api_call) -> decltype(auto) { + return conn->AuthorizedApiCall( + SubApi{[&](ApiContext& api) { + api_call.call(api, sc); + }}); + }, + // ApiRequestHandler + [&](ApiRequestHandler& api_request) -> decltype(auto) { + return conn->AuthorizedApiCall( + SubApi{[&](ApiContext& api) { + api_request(api, sc, this); + }}); + }, + }, + request_); - sr.state_subs.Push( - // if server stream changed its channel + // if request write failed + sr.state_subs += swa.status_event().Subscribe([this, sc](auto status) { + if (status == WriteAction::Status::kFail) { + AE_TELED_WARNING("Request write error"); + OnWriteFailed(sc); + } + }); + // if server stream changed its channel, retry on new channel + sr.state_subs += conn->server_connection().channel_changed_event().Subscribe([this, sc]() { AE_TELED_WARNING("Request server channel changed"); - RemoveRequest(sc); - EnqueueMakeRequest(); - }), - // if server stream is disconnected - conn->server_connection().server_error_event().Subscribe([this, sc]() { - AE_TELED_WARNING("Request server error"); - RemoveRequest(sc); - EnqueueMakeRequest(); - }), - // if request write failed - swa.status_event().Subscribe([this, sc](auto status) { - if (status == WriteAction::Status::kFail) { - AE_TELED_ERROR("Request write error"); - RemoveRequest(sc); - EnqueueMakeRequest(); - } - })); - // TODO: add request server level timeout - if (listener_) { - // update listener in case if servers changed - sr.state_subs += listener_(conn->client_safe_api(), sc, this); + OnChannelChanged(sc); + }); + + // Set per-server request timeout + sr.timeout_sub = ae_context_.scheduler().DelayedTask( + [this, sc]() { + AE_TELED_WARNING("Request timeout for server {}", + sc->server()->server_id); + OnServerRequestTimeout(sc); + }, + request_timeout_); + + if (std::holds_alternative(request_)) { + auto& listener = std::get(request_).listener; + if (listener) { + sr.state_subs += listener(conn->client_safe_api(), sc, this); + } + } +} + +void CloudRequest::OnChannelChanged(CloudServerConnection* sc) { + auto it = server_requests_.find(sc); + if (it == server_requests_.end()) { + return; + } + auto& sr = it->second; + if (sr.retry_count >= max_retries_) { + AE_TELED_WARNING("Server {} retry budget exhausted", + sc->server()->server_id); + sr.exhausted = true; + EnqueueMakeRequest(); + return; + } + // Channel already changed, just re-send. + EnqueueMakeRequest(); +} + +void CloudRequest::OnWriteFailed(CloudServerConnection* sc) { + auto it = server_requests_.find(sc); + if (it == server_requests_.end()) { + return; + } + auto& sr = it->second; + sr.retry_count++; + if (sr.retry_count >= max_retries_) { + AE_TELED_WARNING("Server {} retry budget exhausted on write failure", + sc->server()->server_id); + sr.exhausted = true; + } + EnqueueMakeRequest(); +} + +void CloudRequest::OnServerRequestTimeout(CloudServerConnection* sc) { + auto it = server_requests_.find(sc); + if (it == server_requests_.end()) { + return; + } + auto& sr = it->second; + if (sr.retry_count >= max_retries_) { + AE_TELED_WARNING("Server {} retry budget exhausted on timeout", + sc->server()->server_id); + sr.exhausted = true; + EnqueueMakeRequest(); + return; } + sr.retry_count++; + // Timeout means something is wrong with the stream. + // Restream to switch channels; channel_changed_event will trigger re-send. + sc->Restream(); } -void CloudRequestAction::ServersUpdated() { EnqueueMakeRequest(); } +void CloudRequest::ServersUpdated() { EnqueueMakeRequest(); } -void CloudRequestAction::RemoveRequest( - CloudServerConnection* server_connection) { +void CloudRequest::RemoveRequest(CloudServerConnection* server_connection) { server_requests_.erase(server_connection); } -void CloudRequestAction::EnqueueMakeRequest() { +void CloudRequest::EnqueueMakeRequest() { // enqueue only once at a time if (task_sub_) { return; } - task_sub_.emplace(ae_context_.scheduler().Task([this]() { - task_sub_.reset(); + task_sub_ = ae_context_.scheduler().Task([this]() { + task_sub_.Reset(); MakeRequest(); - })); + }); } -void CloudRequestAction::Finish() { +void CloudRequest::Finish() { swa_sub_.Reset(); server_changed_sub_.Reset(); - task_sub_.reset(); + task_sub_.Reset(); + server_requests_.clear(); Action::Finish(); } diff --git a/aether/cloud_connections/cloud_request.h b/aether/cloud_connections/cloud_request.h index 21915683..ceb6d6f5 100644 --- a/aether/cloud_connections/cloud_request.h +++ b/aether/cloud_connections/cloud_request.h @@ -26,85 +26,58 @@ #include "aether/cloud_connections/cloud_server_connections.h" namespace ae { -class CloudRequest { - class EmptyConnectionsWA final : public WriteAction { - public: - explicit EmptyConnectionsWA(AeContext const& ae_context) { - ae_context.scheduler().Task( - [&]() { WriteAction::SetStatus(WriteAction::Status::kFail); }); - } - }; - - class ReplicaWA final : public WriteAction { - public: - explicit ReplicaWA(std::vector&& swas) noexcept; - - void Stop() noexcept override; - - private: - std::vector swas_; - std::size_t failed_actions_{}; - std::size_t stopped_actions_{}; - MultiSubscription subs_; - }; - - public: - CloudRequest(AeContext const& ae_context, CloudServerConnections& connection); - - WriteAction& CallApi( - AuthApiCaller const& api_caller, - RequestPolicy::Variant policy = RequestPolicy::MainServer{}); - - private: - WriteAction& EmptyWriteAction(); - WriteAction& ReplicaWriteAction(std::vector&& swas); - - AeContext ae_context_; - CloudServerConnections* connection_; - std::optional empty_wa_; - std::vector replica_was_; -}; - /** * \brief Makes request according to the request policy. * If request fails or times out, it will restream the cloud connection and - * retrie until max_attempts is reached. - * ResponseListener listener must subscribe to client_api and handle the - * response. On success, listener must call CloudRequestAction::Succeeded(). On - * failure, listener must call CloudRequestAction::Failed(). + * retry on different channels. When a server exhausts its retry budget, + * it moves on to the next server in the list. + * ResponseSubscriber must subscribe to client_api and handle the + * response. On success, listener must call CloudRequest::Succeeded(). On + * failure, listener must call CloudRequest::Failed(). */ -class CloudRequestAction final : public Action { +class CloudRequest final : public Action { struct ServerRequest { MultiSubscription state_subs; - bool is_active{false}; + TaskSubscription timeout_sub; + std::size_t retry_count{0}; + bool exhausted{false}; }; public: - using SuccessEvent = Event; - using FailureEvent = Event; + static constexpr std::size_t kDefaultMaxRetries = 5; + static constexpr Duration kDefaultRequestTimeout = + std::chrono::milliseconds{AE_CLOUD_REQUEST_TIMEOUT_MS}; + + using ResultEvent = Event; - CloudRequestAction(AeContext const& ae_context, AuthApiCaller&& api_caller, - ClientResponseListener&& listener, - CloudServerConnections& cloud_server_connections, - RequestPolicy::Variant policy); + CloudRequest(AeContext const& ae_context, ApiCallWithListener&& api_call, + CloudServerConnections& cloud_server_connections, + RequestPolicy::Variant policy, + std::size_t max_retries = kDefaultMaxRetries, + Duration request_timeout = kDefaultRequestTimeout); - CloudRequestAction(AeContext const& ae_context, AuthApiRequest&& api_request, - CloudServerConnections& cloud_server_connections, - RequestPolicy::Variant policy); + CloudRequest(AeContext const& ae_context, ApiRequestHandler&& api_request, + CloudServerConnections& cloud_server_connections, + RequestPolicy::Variant policy, + std::size_t max_retries = kDefaultMaxRetries, + Duration request_timeout = kDefaultRequestTimeout); - AE_CLASS_NO_COPY_MOVE(CloudRequestAction) + AE_CLASS_NO_COPY_MOVE(CloudRequest) void Succeeded(); void Failed(); - SuccessEvent::Subscriber success_event(); - FailureEvent::Subscriber failure_event(); + ResultEvent::Subscriber result_event(); private: void MakeRequest(); void MakeServerRequest(CloudServerConnection* sc, ServerRequest& sr); + void PrefillServerRequests(); void ServersUpdated(); + void OnChannelChanged(CloudServerConnection* sc); + void OnServerRequestTimeout(CloudServerConnection* sc); + void OnWriteFailed(CloudServerConnection* sc); void RemoveRequest(CloudServerConnection* server_connection); void EnqueueMakeRequest(); @@ -112,16 +85,16 @@ class CloudRequestAction final : public Action { void Finish(); AeContext ae_context_; - std::variant request_; - ClientResponseListener listener_; - CloudServerConnections* cloud_sc_; + std::variant request_; + CloudServerConnections* cloud_scs_; RequestPolicy::Variant policy_; - std::optional task_sub_; + std::size_t max_retries_; + Duration request_timeout_; + TaskSubscription task_sub_; Subscription swa_sub_; Subscription server_changed_sub_; - SuccessEvent success_event_; - FailureEvent failure_event_; + ResultEvent result_event_; std::map server_requests_; }; diff --git a/aether/cloud_connections/cloud_server_connections.cpp b/aether/cloud_connections/cloud_server_connections.cpp index 6e702673..576b97d8 100644 --- a/aether/cloud_connections/cloud_server_connections.cpp +++ b/aether/cloud_connections/cloud_server_connections.cpp @@ -16,13 +16,53 @@ #include "aether/cloud_connections/cloud_server_connections.h" +#include + #include "aether/server.h" +#include "aether/api_protocol/api_protocol.h" #include "aether/server_connections/server_connection.h" #include "aether/tele/tele.h" namespace ae { +cloud_server_connections_internal::EmptyConnectionsWA::EmptyConnectionsWA( + AeContext const& ae_context) { + ae_context.scheduler().Task( + [&]() { WriteAction::SetStatus(WriteAction::Status::kFail); }); +} + +cloud_server_connections_internal::ReplicaWA::ReplicaWA( + std::vector&& swas) noexcept + : swas_{std::move(swas)} { + assert(!swas_.empty()); + for (auto* action : swas_) { + subs_ += + action->status_event().Subscribe([this](WriteAction::Status status) { + switch (status) { + case WriteAction::Status::kSuccess: + SetStatus(status); + return; + case WriteAction::Status::kFail: + failed_actions_++; + break; + case WriteAction::Status::kStop: + stopped_actions_++; + break; + } + if ((failed_actions_ + stopped_actions_) >= swas_.size()) { + SetStatus(status); + } + }); + } +} + +void cloud_server_connections_internal::ReplicaWA::Stop() noexcept { + for (auto* action : swas_) { + action->Stop(); + } +} + CloudServerConnections::CloudServerConnections( AeContext const& ae_context, Ptr const& cloud, std::unique_ptr connection_factory, @@ -40,11 +80,15 @@ CloudServerConnections::servers_update_event() { return servers_update_event_; } -std::vector const& CloudServerConnections::servers() +std::vector const& CloudServerConnections::selected_servers() const { return selected_servers_; } +std::vector const& CloudServerConnections::servers() { + return all_servers_; +} + std::size_t CloudServerConnections::count_connections() const { return selected_servers_.size(); } @@ -62,9 +106,17 @@ void CloudServerConnections::Restream() { void CloudServerConnections::InitServerConnections() { server_connections_.clear(); - for (auto& server : cloud_->servers()) { + auto cloud = cloud_.Lock(); + assert(cloud); + server_connections_.reserve(cloud->servers().size()); + for (auto& server : cloud->servers()) { server_connections_.emplace_back(server.Load(), *connection_factory_); } + all_servers_.clear(); + all_servers_.reserve(server_connections_.size()); + for (auto& s : server_connections_) { + all_servers_.emplace_back(&s); + } } void CloudServerConnections::InitServers() { @@ -82,13 +134,17 @@ void CloudServerConnections::SelectServers( std::vector const& servers) { auto select_count = std::min(servers.size(), max_connections_); - auto get_ids = [&](auto const& ss) { + auto get_ids = [&]([[maybe_unused]] auto const& ss) noexcept { +#if DEBUG std::vector sids; sids.reserve(ss.size()); for (auto const* server : ss) { sids.emplace_back(server->server()->server_id); } return sids; +#else + return "!not debug!"; +#endif }; AE_TELED_DEBUG("Select servers count {} from sids [{}]", select_count, @@ -156,19 +212,23 @@ void CloudServerConnections::SubscribeToServerState( void CloudServerConnections::ReselectServers() { // reselct servers on next cycle - defer_sub_ = ae_context_.scheduler().Task([&]() { InitServers(); }); + if (defer_sub_) { + return; + } + defer_sub_ = ae_context_.scheduler().Task([&]() { + defer_sub_.Reset(); + InitServers(); + }); } void CloudServerConnections::UnselectServer( CloudServerConnection& server_connection) { - auto it = std::find(std::begin(selected_servers_), - std::end(selected_servers_), &server_connection); - if (it == std::end(selected_servers_)) { - return; + auto old_size = selected_servers_.size(); + std::erase(selected_servers_, &server_connection); + if (selected_servers_.size() < old_size) { + AE_TELED_DEBUG("Servers unselected, remaining count {}", + selected_servers_.size()); } - it = selected_servers_.erase(it); - AE_TELED_DEBUG("Servers unselected, remaining count {}", - selected_servers_.size()); } void CloudServerConnections::QuarantineTimer( @@ -213,4 +273,38 @@ std::vector CloudServerConnections::ServerCandidates() { return servers; } +WriteAction& CloudServerConnections::CallApi(ApiCall const& api_caller, + RequestPolicy::Variant policy) { + std::vector swas; + ForServers( + [&](CloudServerConnection* sc) { + auto* conn = sc->client_connection(); + assert((conn != nullptr) && "Client connection is null"); + swas.emplace_back(&conn->AuthorizedApiCall( + SubApi{[&](auto& api) { api_caller(api, sc); }})); + }, + policy); + + if (swas.empty()) { + return EmptyWriteAction(); + } + if (swas.size() == 1) { + return *swas.front(); + } + return ReplicaWriteAction(std::move(swas)); +} + +WriteAction& CloudServerConnections::EmptyWriteAction() { + if (!empty_wa_ || empty_wa_->is_finished()) { + empty_wa_.emplace(ae_context_); + } + return *empty_wa_; +} + +WriteAction& CloudServerConnections::ReplicaWriteAction( + std::vector&& swas) { + // FIXME: this vector only grows + return replica_was_.emplace_back(std::move(swas)); +} + } // namespace ae diff --git a/aether/cloud_connections/cloud_server_connections.h b/aether/cloud_connections/cloud_server_connections.h index 2ea28c53..4ba70924 100644 --- a/aether/cloud_connections/cloud_server_connections.h +++ b/aether/cloud_connections/cloud_server_connections.h @@ -16,21 +16,45 @@ #ifndef AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTIONS_H_ #define AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTIONS_H_ +#include #include #include +#include #include "aether/cloud.h" #include "aether/ptr/ptr.h" +#include "aether/ptr/ptr_view.h" #include "aether/ae_context.h" #include "aether/events/events.h" +#include "aether/events/multi_subscription.h" +#include "aether/write_action/write_action.h" +#include "aether/cloud_connections/request_policy.h" +#include "aether/cloud_connections/cloud_callbacks.h" #include "aether/cloud_connections/cloud_server_connection.h" #include "aether/server_connections/iserver_connection_factory.h" namespace ae { -class CloudServerConnections { - friend class CloudRequest; +namespace cloud_server_connections_internal { +class EmptyConnectionsWA final : public WriteAction { + public: + explicit EmptyConnectionsWA(AeContext const& ae_context); +}; + +class ReplicaWA final : public WriteAction { + public: + explicit ReplicaWA(std::vector&& swas) noexcept; + void Stop() noexcept override; + private: + std::vector swas_; + std::size_t failed_actions_{}; + std::size_t stopped_actions_{}; + MultiSubscription subs_; +}; +} // namespace cloud_server_connections_internal + +class CloudServerConnections { public: using ServersUpdate = Event; @@ -46,7 +70,11 @@ class CloudServerConnections { /** * \brief List of currently selected servers in priority order */ - std::vector const& servers() const; + std::vector const& selected_servers() const; + /** + * \brief List of all server connections including quarantined ones. + */ + std::vector const& servers(); std::size_t count_connections() const; std::size_t max_connections() const; @@ -56,7 +84,61 @@ class CloudServerConnections { */ void Restream(); + /** + * \brief Iterate over servers according to the request policy. + * Calls func with CloudServerConnection* for each server. + */ + template + void ForServers(TFunc&& func, + RequestPolicy::Variant policy = RequestPolicy::MainServer{}) { + std::visit([&](auto p) { ForServersImpl(std::forward(func), p); }, + policy); + } + + /** + * \brief Fire-and-forget API call to servers per policy. + * Returns a WriteAction that reflects the write status. + */ + WriteAction& CallApi( + ApiCall const& api_caller, + RequestPolicy::Variant policy = RequestPolicy::MainServer{}); + private: + template + void ForServersImpl(TFunc&& func, RequestPolicy::MainServer) { + if (selected_servers_.empty()) { + return; + } + std::invoke(std::forward(func), selected_servers_.front()); + } + + template + void ForServersImpl(TFunc&& func, RequestPolicy::Priority priority) { + if (priority.priority >= selected_servers_.size()) { + return; + } + std::invoke(std::forward(func), + selected_servers_.at(priority.priority)); + } + + template + void ForServersImpl(TFunc&& func, RequestPolicy::Replica replica) { + auto visit_count = std::min(selected_servers_.size(), replica.count); + for (std::size_t i = 0; i < visit_count; ++i) { + std::invoke(std::forward(func), selected_servers_.at(i)); + } + } + + template + void ForServersImpl(TFunc&& func, RequestPolicy::All) { + for (auto* sc : selected_servers_) { + std::invoke(std::forward(func), sc); + } + } + + WriteAction& EmptyWriteAction(); + WriteAction& ReplicaWriteAction(std::vector&& swas); + void InitServerConnections(); void InitServers(); void SelectServers(std::vector const& servers); @@ -69,11 +151,12 @@ class CloudServerConnections { std::vector ServerCandidates(); AeContext ae_context_; - Ptr cloud_; + PtrView cloud_; std::unique_ptr connection_factory_; std::size_t max_connections_; std::vector server_connections_; + std::vector all_servers_; // selected list of servers sorted by the priority std::vector selected_servers_; @@ -81,6 +164,10 @@ class CloudServerConnections { std::map server_state_subs_; TaskSubscription defer_sub_; + + std::optional + empty_wa_; + std::vector replica_was_; }; } // namespace ae diff --git a/aether/cloud_connections/cloud_subscription.cpp b/aether/cloud_connections/cloud_subscription.cpp index 0323bbaa..6ef1797c 100644 --- a/aether/cloud_connections/cloud_subscription.cpp +++ b/aether/cloud_connections/cloud_subscription.cpp @@ -16,33 +16,31 @@ #include "aether/cloud_connections/cloud_subscription.h" -#include "aether/cloud_connections/cloud_visit.h" - namespace ae { -CloudSubscription::CloudSubscription(ClientListener subscriber, - CloudServerConnections& cloud_connection, - RequestPolicy::Variant request_policy) +CloudEventListener::CloudEventListener(ApiEventSubscriber subscriber, + CloudServerConnections& cloud_connection, + RequestPolicy::Variant request_policy) : cloud_connection_{&cloud_connection}, subscriber_{std::move(subscriber)}, request_policy_{request_policy} { server_update_sub_ = cloud_connection_->servers_update_event().Subscribe( - MethodPtr<&CloudSubscription::ServersUpdate>{this}); + MethodPtr<&CloudEventListener::ServersUpdate>{this}); ServersUpdate(); } -CloudSubscription::CloudSubscription(CloudSubscription&& other) noexcept +CloudEventListener::CloudEventListener(CloudEventListener&& other) noexcept : cloud_connection_{other.cloud_connection_}, subscriber_{std::move(other.subscriber_)}, request_policy_{other.request_policy_}, subscriptions_{std::move(other.subscriptions_)} { if (cloud_connection_ != nullptr) { server_update_sub_ = cloud_connection_->servers_update_event().Subscribe( - MethodPtr<&CloudSubscription::ServersUpdate>{this}); + MethodPtr<&CloudEventListener::ServersUpdate>{this}); } } -CloudSubscription& CloudSubscription::operator=( - CloudSubscription&& other) noexcept { +CloudEventListener& CloudEventListener::operator=( + CloudEventListener&& other) noexcept { if (this != &other) { cloud_connection_ = other.cloud_connection_; subscriber_ = std::move(other.subscriber_); @@ -50,25 +48,25 @@ CloudSubscription& CloudSubscription::operator=( subscriptions_ = std::move(other.subscriptions_); if (cloud_connection_ != nullptr) { server_update_sub_ = cloud_connection_->servers_update_event().Subscribe( - MethodPtr<&CloudSubscription::ServersUpdate>{this}); + MethodPtr<&CloudEventListener::ServersUpdate>{this}); } } return *this; } -void CloudSubscription::ServersUpdate() { +void CloudEventListener::ServersUpdate() { // clean old subscriptions and make new subscriptions_.Reset(); - CloudVisit::Visit( + cloud_connection_->ForServers( [&](auto* sc) { auto* conn = sc->client_connection(); assert((conn != nullptr) && "ClientConnection is null"); subscriptions_ += subscriber_(conn->client_safe_api(), sc); }, - *cloud_connection_, request_policy_); + request_policy_); } -void CloudSubscription::Reset() { +void CloudEventListener::Reset() { server_update_sub_.Reset(); subscriptions_.Reset(); } diff --git a/aether/cloud_connections/cloud_subscription.h b/aether/cloud_connections/cloud_subscription.h index d23a8767..a20a0621 100644 --- a/aether/cloud_connections/cloud_subscription.h +++ b/aether/cloud_connections/cloud_subscription.h @@ -14,8 +14,8 @@ * limitations under the License. */ -#ifndef AETHER_CLOUD_CONNECTIONS_CLOUD_SUBSCRIPTION_H_ -#define AETHER_CLOUD_CONNECTIONS_CLOUD_SUBSCRIPTION_H_ +#ifndef AETHER_CLOUD_CONNECTIONS_CLOUD_EVENT_LISTENER_H_ +#define AETHER_CLOUD_CONNECTIONS_CLOUD_EVENT_LISTENER_H_ #include "aether/common.h" @@ -24,17 +24,17 @@ #include "aether/cloud_connections/cloud_server_connections.h" namespace ae { -class CloudSubscription { +class CloudEventListener { public: - CloudSubscription() = default; - CloudSubscription( - ClientListener subscriber, CloudServerConnections& cloud_connection, + CloudEventListener() = default; + CloudEventListener( + ApiEventSubscriber subscriber, CloudServerConnections& cloud_connection, RequestPolicy::Variant request_policy = RequestPolicy::MainServer{}); - CloudSubscription(CloudSubscription&& other) noexcept; - CloudSubscription& operator=(CloudSubscription&& other) noexcept; + CloudEventListener(CloudEventListener&& other) noexcept; + CloudEventListener& operator=(CloudEventListener&& other) noexcept; - AE_CLASS_NO_COPY(CloudSubscription) + AE_CLASS_NO_COPY(CloudEventListener) void Reset(); @@ -42,11 +42,11 @@ class CloudSubscription { void ServersUpdate(); CloudServerConnections* cloud_connection_{}; - ClientListener subscriber_; + ApiEventSubscriber subscriber_; RequestPolicy::Variant request_policy_; Subscription server_update_sub_; MultiSubscription subscriptions_; }; } // namespace ae -#endif // AETHER_CLOUD_CONNECTIONS_CLOUD_SUBSCRIPTION_H_ +#endif // AETHER_CLOUD_CONNECTIONS_CLOUD_EVENT_LISTENER_H_ diff --git a/aether/cloud_connections/cloud_visit.h b/aether/cloud_connections/cloud_visit.h deleted file mode 100644 index df9a4d63..00000000 --- a/aether/cloud_connections/cloud_visit.h +++ /dev/null @@ -1,79 +0,0 @@ -/* - * Copyright 2025 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_CLOUD_CONNECTIONS_CLOUD_VISIT_H_ -#define AETHER_CLOUD_CONNECTIONS_CLOUD_VISIT_H_ - -#include "aether/cloud_connections/request_policy.h" -#include "aether/cloud_connections/cloud_server_connections.h" - -namespace ae { -class CloudVisit { - public: - template - static void Visit( - TFunc&& func, CloudServerConnections& cloud_connection, - RequestPolicy::Variant policy = RequestPolicy::MainServer{}) { - std::visit( - [&](auto p) { - VisitImpl(std::forward(func), cloud_connection, p); - }, - policy); - } - - private: - template - static void VisitImpl(TFunc&& func, CloudServerConnections& cloud_connection, - RequestPolicy::MainServer) { - if (cloud_connection.servers().empty()) { - return; - } - auto* sc = cloud_connection.servers().front(); - std::invoke(std::forward(func), sc); - } - - template - static void VisitImpl(TFunc&& func, CloudServerConnections& cloud_connection, - RequestPolicy::Priority priority) { - if (priority.priority >= cloud_connection.servers().size()) { - return; - } - auto* sc = cloud_connection.servers().at(priority.priority); - std::invoke(std::forward(func), sc); - } - - template - static void VisitImpl(TFunc&& func, CloudServerConnections& cloud_connection, - RequestPolicy::Replica replica) { - auto visit_count = - std::min(cloud_connection.servers().size(), replica.count); - for (std::size_t i = 0; i < visit_count; ++i) { - auto* sc = cloud_connection.servers().at(i); - std::invoke(std::forward(func), sc); - } - } - - template - static void VisitImpl(TFunc&& func, CloudServerConnections& cloud_connection, - RequestPolicy::All) { - for (auto* sc : cloud_connection.servers()) { - std::invoke(std::forward(func), sc); - } - } -}; -} // namespace ae - -#endif // AETHER_CLOUD_CONNECTIONS_CLOUD_VISIT_H_ diff --git a/aether/config.h b/aether/config.h index 9e6c9e9e..220927a1 100644 --- a/aether/config.h +++ b/aether/config.h @@ -302,6 +302,11 @@ # define AE_CLOUD_SERVER_QUARANTINE_TIME_MS 10000 #endif +// Cloud request per-server timeout in milliseconds +#ifndef AE_CLOUD_REQUEST_TIMEOUT_MS +# define AE_CLOUD_REQUEST_TIMEOUT_MS 5000 +#endif + // Time synchronization enabled #ifndef AE_TIME_SYNC_ENABLED # define AE_TIME_SYNC_ENABLED 1 diff --git a/aether/connection_manager/client_cloud_manager.cpp b/aether/connection_manager/client_cloud_manager.cpp index c21f4e2a..e9c284d8 100644 --- a/aether/connection_manager/client_cloud_manager.cpp +++ b/aether/connection_manager/client_cloud_manager.cpp @@ -210,8 +210,8 @@ void ClientCloudManager::ListenForCloudUpdate() { auto client = Client::ptr{client_}.Load(); assert(client != nullptr && "Client does not loaded"); - cloud_update_sub_ = CloudSubscription{ - ClientListener{[this](ClientApiSafe& client_api, + cloud_update_sub_ = CloudEventListener{ + ApiEventSubscriber{[this](ClientApiSafe& client_api, CloudServerConnection* /*server_connection*/) { return client_api.send_cloud_configs().Subscribe( MethodPtr<&ClientCloudManager::CloudConfigs>{this}); diff --git a/aether/connection_manager/client_cloud_manager.h b/aether/connection_manager/client_cloud_manager.h index 1fb06c33..c1d719cf 100644 --- a/aether/connection_manager/client_cloud_manager.h +++ b/aether/connection_manager/client_cloud_manager.h @@ -113,7 +113,7 @@ class ClientCloudManager : public Obj { std::map cloud_cache_; CloudUpdateEvent cloud_update_event_; - CloudSubscription cloud_update_sub_; + CloudEventListener cloud_update_sub_; std::optional cloud_actions_; std::optional get_servers_pool_; std::vector{this})}, - cloud_request_{ae_context_, cloud_connection} { + MethodPtr<&GetCloudFromAether::CloudUpdate>{this})} { RequestCloud(); } @@ -41,19 +41,18 @@ GetCloudFromAether::result_event() noexcept { void GetCloudFromAether::RequestCloud() { AE_TELED_DEBUG("RequestCloud"); - cloud_request_.CallApi( - AuthApiCaller{[&](ApiContext& auth_api, + cloud_connection_.CallApi( + ApiCall{[&](ApiContext& auth_api, CloudServerConnection* server_connection) { AE_TELED_DEBUG("Send cloud request for uid:{} at server:{}", client_uid_, server_connection->server()->server_id); auth_api->report_applied_config(std::vector{AppliedConfig{ .subject_uid = client_uid_, - .config_version = -1, // -1 means request config + .config_version = -1, }}); }}, RequestPolicy::All{}); - // response shall be received through CloudUpdate by matching with uid } void GetCloudFromAether::CloudUpdate( diff --git a/aether/connection_manager/get_cloud_aether.h b/aether/connection_manager/get_cloud_aether.h index 78fcd523..20c03b3e 100644 --- a/aether/connection_manager/get_cloud_aether.h +++ b/aether/connection_manager/get_cloud_aether.h @@ -21,7 +21,6 @@ #include "aether/ae_context.h" #include "aether/events/events.h" -#include "aether/cloud_connections/cloud_request.h" #include "aether/connection_manager/get_cloud_action.h" #include "aether/cloud_connections/cloud_server_connections.h" @@ -44,9 +43,9 @@ class GetCloudFromAether final : public GetCloudAction { AeContext ae_context_; Uid client_uid_; + CloudServerConnections& cloud_connection_; Subscription cloud_update_sub_; - CloudRequest cloud_request_; ResultEvent result_event_; }; } // namespace ae From bdc0c5f7665a224f6fedc920c804e3221fb0c667 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Thu, 25 Jun 2026 17:58:46 +0500 Subject: [PATCH 5/6] use cloud request in get cloud aether --- .../connection_manager/get_cloud_aether.cpp | 45 +++++++++++-------- aether/connection_manager/get_cloud_aether.h | 5 ++- 2 files changed, 30 insertions(+), 20 deletions(-) diff --git a/aether/connection_manager/get_cloud_aether.cpp b/aether/connection_manager/get_cloud_aether.cpp index 5825091e..923dbb27 100644 --- a/aether/connection_manager/get_cloud_aether.cpp +++ b/aether/connection_manager/get_cloud_aether.cpp @@ -28,9 +28,34 @@ GetCloudFromAether::GetCloudFromAether(AeContext const& ae_context, : ae_context_{ae_context}, client_uid_{client_uid}, cloud_connection_{cloud_connection}, + cloud_request_{ + ae_context, + ApiCallWithListener{ + ApiCall{[this](ApiContext& auth_api, + CloudServerConnection* server_connection) { + AE_TELED_DEBUG("Send cloud request for uid:{} at server:{}", + client_uid_, + server_connection->server()->server_id); + auth_api->report_applied_config(std::vector{AppliedConfig{ + .subject_uid = client_uid_, + .config_version = -1, + }}); + }}, + // listen on CloudUpdate + ResponseSubscriber{}}, + cloud_connection_, + RequestPolicy::All{}, + }, cloud_update_sub_{client_cloud_manager.cloud_update_event().Subscribe( MethodPtr<&GetCloudFromAether::CloudUpdate>{this})} { - RequestCloud(); + cloud_request_result_sub_ = cloud_request_.result_event().Subscribe( + [this](bool success) { + if (!success) { + AE_TELED_ERROR("CloudRequest failed for uid:{}", client_uid_); + result_event_.Emit(Error{-1}); + Finish(); + } + }); } GetCloudFromAether::ResultEvent::Subscriber @@ -38,23 +63,6 @@ GetCloudFromAether::result_event() noexcept { return EventSubscriber{result_event_}; } -void GetCloudFromAether::RequestCloud() { - AE_TELED_DEBUG("RequestCloud"); - - cloud_connection_.CallApi( - ApiCall{[&](ApiContext& auth_api, - CloudServerConnection* server_connection) { - AE_TELED_DEBUG("Send cloud request for uid:{} at server:{}", - client_uid_, server_connection->server()->server_id); - - auth_api->report_applied_config(std::vector{AppliedConfig{ - .subject_uid = client_uid_, - .config_version = -1, - }}); - }}, - RequestPolicy::All{}); -} - void GetCloudFromAether::CloudUpdate( Uid const& uid, Result const& res) { if (uid != client_uid_) { @@ -66,6 +74,7 @@ void GetCloudFromAether::CloudUpdate( } else { result_event_.Emit(Error{res.error()}); } + cloud_request_.Succeeded(); } } // namespace ae diff --git a/aether/connection_manager/get_cloud_aether.h b/aether/connection_manager/get_cloud_aether.h index 20c03b3e..6b6223d3 100644 --- a/aether/connection_manager/get_cloud_aether.h +++ b/aether/connection_manager/get_cloud_aether.h @@ -22,6 +22,7 @@ #include "aether/events/events.h" #include "aether/connection_manager/get_cloud_action.h" +#include "aether/cloud_connections/cloud_request.h" #include "aether/cloud_connections/cloud_server_connections.h" namespace ae { @@ -38,13 +39,13 @@ class GetCloudFromAether final : public GetCloudAction { ResultEvent::Subscriber result_event() noexcept override; private: - void RequestCloud(); void CloudUpdate(Uid const& uid, Result const& res); AeContext ae_context_; Uid client_uid_; CloudServerConnections& cloud_connection_; - + CloudRequest cloud_request_; + Subscription cloud_request_result_sub_; Subscription cloud_update_sub_; ResultEvent result_event_; }; From a04ccb26883ffd713f1baf55fcc99c6c04aa2a74 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Thu, 25 Jun 2026 17:59:11 +0500 Subject: [PATCH 6/6] add asserts to test if empty data is send into tcp and udp --- aether/transport/system_sockets/tcp/tcp.h | 2 ++ aether/transport/system_sockets/udp/udp.h | 1 + 2 files changed, 3 insertions(+) diff --git a/aether/transport/system_sockets/tcp/tcp.h b/aether/transport/system_sockets/tcp/tcp.h index d88612b0..ff8f18d4 100644 --- a/aether/transport/system_sockets/tcp/tcp.h +++ b/aether/transport/system_sockets/tcp/tcp.h @@ -169,6 +169,8 @@ class TcpTransport final : public tcp_internal::TcpBase { } WriteAction& Write(DataBuffer&& in_data) override { + assert(in_data.size() != 0); + AE_TELE_DEBUG(kTcpTransportSend, "Socket {} send data size {}", endpoint_, in_data.size()); diff --git a/aether/transport/system_sockets/udp/udp.h b/aether/transport/system_sockets/udp/udp.h index 1f52e1c3..9e574c9b 100644 --- a/aether/transport/system_sockets/udp/udp.h +++ b/aether/transport/system_sockets/udp/udp.h @@ -157,6 +157,7 @@ class UdpTransport final : public upd_internal::UdpBase { } WriteAction& Write(DataBuffer&& in_data) override { + assert(in_data.size() != 0); AE_TELE_DEBUG(kUdpTransportSend, "Socket {} send data size:{}", endpoint_, in_data.size()); auto* send_action =