From 249d0b762690bda58b6880657ed91bf09a3eff69 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Wed, 19 Aug 2026 19:22:21 +0500 Subject: [PATCH 1/3] update AGENTSmd with serialization and reflection --- AGENTS.md | 246 +++++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 218 insertions(+), 28 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 0a8f5d0c..96cfafca 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,27 +1,5 @@ # Aether Client C++ Guide -## Modes and Persistence - -`aether` has three persistence behaviors: - -- **Distillation** (`AE_DISTILLATION=On`): create every object from scratch, - even when persistent state already exists. Use it during development and as - preparation for production. -- **Filtration** (`AE_FILTRATION=On`): load an object when state exists and - create it when it does not. Filtration also enables the distillation code - paths at compile time. It is useful when applications should tolerate both - existing and missing state. -- **Production**: both `AE_DISTILLATION` and `AE_FILTRATION` are disabled or - undefined. Main persistent objects, such as `Aether`, adapters, clients, and - clouds, must already exist. Some argument-taking constructors are disabled; - objects must be loaded from the domain or copied from prefab objects. - -Production is descriptive terminology, not a separate build option. - -`FS_INIT` may provide generated or static persisted-state maps. Persistence is -not implied by mutation; use the established application save path when state -must survive shutdown. - ## Project Model `aether` is a C++20 static library for persistent state, asynchronous actions @@ -43,6 +21,10 @@ persistent `Obj` types. and implement the established `Load`/`Save` patterns. - Use `ae::ObjPtr` for strong references to persistent objects. - Use `ae::Ptr` for shared ownership of non-`Obj` objects. +- `ae::Ptr` uses reachability counting on the reference graph to reclaim + cyclic references. Releasing a pointer may therefore be expensive; avoid + unnecessary copies, pass it by reference when possible, and move it when + transferring ownership. - `ae::PtrView` is a weak, nullable view. Lock/load it before retaining or dereferencing the object. - A valid `ObjPtr` may still refer to an unloaded object. Load it before use @@ -178,6 +160,93 @@ ordinary errors. - Use `SubApi` and the existing API context/parser patterns for nested API calls instead of inventing a separate packet format. +## Serialization + +Serialization is provided by the `aether-miscpp` dependency. It is used both +to save and load persistent `Obj` state and to encode and decode API protocol +messages. + +There are three ways to make a type serializable: + +- Use the project's reflection support when the type is a straightforward + aggregate of serializable members. +- Provide a `seri::Serializer` specialization. Prefer this + approach because it keeps serialization logic separate from the model type. +- Add `Seri()` and `Deseri()` member functions when serialization intrinsically + belongs to the type or a member serializer is otherwise the best fit. + +Implement a serializer against the general `seri::Archive` concept when the +representation is independent of the underlying archive. Specialize it for a +specific archive when the representation depends on that archive's storage or +wire format. The currently available concrete archive is +`seri::BinaryArchive`. + +A serializer provides `Seri()` for saving and `Deseri()` for loading. Saving +receives `Meta` and loading receives `Meta`; both return +`SeriResult`: + +```cpp +namespace ae::seri { +template +struct Serializer { + SeriResult Seri(A& archive, Meta meta) const { + return archive.Save(Meta{meta.value.member}); + } + + SeriResult Deseri(A& archive, Meta meta) const { + return archive.Load(Meta{meta.value.member}); + } +}; +} // namespace ae::seri +``` + +`BinaryBuffer` exposes two pairs of `Read` and `Write` operations. The size +operation represents a container size, meaning a count of elements. The data +and size operation represents a payload together with its size, for one or +more elements. Different buffer implementations may use different physical +representations for the size and data, so serializers should use the buffer +operations rather than assuming a particular layout. + +## Reflection + +Reflection is provided by the `aether-miscpp` dependency. It describes the +members of a type so generic code can inspect or process them, including +serialization and other algorithms. + +For regular members, declare the reflected members in the type with: + +```cpp +AE_REFLECT_MEMBERS(a, b, c) +``` + +For explicit reflection entries, use `AE_REFLECT` and pass it reflection +entries. Use `AE_MMBR(member)` for one regular member or +`AE_MMBRS(first, second)` for multiple regular members. Use `AE_REF(member)` +when a reflected member is a reference. For example, use +`AE_REFLECT(AE_REF(member))` for a single explicit reference member. For base +classes, `AE_REFLECT` supports both `AE_REF_BASE(Base)` and `AE_BASE(Base)`: + +- Use `AE_REF_BASE(Base)` to reflect a reference to `Base` as one member. +- Use `AE_BASE(Base)` to concatenate `Base`'s reflected members into the + derived type's reflection. + +All explicit reflection helpers can be combined in one declaration: + +```cpp +AE_REFLECT(AE_MMBR(member), AE_MMBRS(first, second), AE_REF(reference), + AE_REF_BASE(BaseAsMember), AE_BASE(BaseMembers)); +``` + +Create a reflection object with `ae::make_reflection(obj)` and apply a +callable to all reflected members with `Apply()`: + +```cpp +auto reflection = ae::make_reflection(obj); +reflection.Apply([](auto&&... members) { + // Process the reflected members. +}); +``` + ## Streams - Streams publish state and data through events; writes return actions. @@ -204,6 +273,38 @@ configured through `aether/tele.h`. - Register a module tag when tagged logging is needed. - Use registered tags with `AE_TELE_(kTag, ...)`. +### Format + +`Format` is provided by the `aether-miscpp` dependency and can be used on its +own to build formatted strings or to provide a format string to a telemetry +log, for example: + +```cpp +AE_TELE_DEBUG(kTag, "Format string {}", data); +``` + +- Use `{}` for replacement fields. Arguments are consumed from left to right; + for example, `Format("id={}, state={}", id, state)`. Formatting schemes can + be selected after a colon, such as `{:time}` for time values. +- To make a project type formattable, specialize `ae::Formatter` and + implement `Format(YourType const&, FormatContext&) const`. Write + output through `ctx.out()`, or delegate to existing formatters with + `Formatter{}.Format(value, ctx)`. For a composed representation, use + `FormatTo(ctx.out(), FormatScheme{"value={}, count={}"}, value, count)`: + + ```cpp + namespace ae { + template <> + struct Formatter { + template + void Format(MyType const& value, FormatContext& ctx) const { + FormatTo(ctx.out(), FormatScheme{"name={}, count={}"}, value.name, + value.count); + } + }; + } // namespace ae + ``` + ## C++ Coding Rules - Follow the Google C++ Style Guide. @@ -271,13 +372,102 @@ explicitly requested to prove specific behavior. ## Build and Configuration -Use the regular root CMake project, enable the required `AE_BUILD_*` options, -build the requested targets, and run their tests from the same build directory. -Run clang-tidy on changed C++ files using that build's matching -`compile_commands.json`; regenerate it when configuration flags change. +Use the regular root CMake project. Keep separate build directories for +different compilers, build types, sanitizers, persistence modes, and user +configuration headers. A configured build directory retains its CMake options, +so inspect or reconfigure it before relying on its settings. + +Use a separate build directory such as `` for each compiler, +platform, build type, sanitizer, persistence mode, or user configuration. +Build and test it with: + +```bash +cmake --build --parallel +ctest --test-dir --output-on-failure +``` + +A successful CMake configure is not build or test validation. + +### Compile-Time Configuration + +`aether/config.h` provides the built-in configuration defaults. `USER_CONFIG` +is optional; when defined, `aether/config.h` includes the selected header before +applying its remaining `#ifndef` defaults. Therefore a user configuration header +overrides the defaults by defining the relevant `AE_*` macros. + +No user configuration is selected when `USER_CONFIG` is empty. This is the +project's default behavior and uses the values from `aether/config.h`. + +Select one of the predefined configurations with a path relative to the source +tree, for example: + +```bash +cmake -S . -B build-hydrogen \ + -DUSER_CONFIG=config/user_config_hydrogen.h \ + -DAE_BUILD_TESTS=ON +``` + +Predefined configurations are located in `config/`. Inspect the selected +configuration before changing code that depends on compile-time feature or +cryptography settings. + +Custom configuration headers may also be supplied through CMake: + +```bash +cmake -S . -B build-custom \ + -DUSER_CONFIG=/absolute/path/to/my_aether_config.h +``` + +Configuration changes require a separate build directory or a CMake reconfigure, +and can change available source features and required platform dependencies. + +`USER_CONFIG` is a compile-time configuration header, not persisted state. +`FS_INIT` optionally supplies generated or static saved-state data: + +```bash +cmake -S . -B build-with-state \ + -DUSER_CONFIG=config/user_config_hydrogen.h \ + -DFS_INIT=/absolute/path/to/generated_state.h +``` + +### Persistence Build Modes + +`AE_DISTILLATION` and `AE_FILTRATION` are independent CMake options: + +- `AE_DISTILLATION=ON` enables creation of objects from scratch, even when + persisted state exists. +- `AE_FILTRATION=ON` enables loading existing state and creating missing + objects. In `aether/config.h`, filtration also defines `AE_DISTILLATION=1` + so code requiring distillation support is compiled. +- With both options disabled, production behavior is used: required persistent + objects must already exist and be loaded from the domain or copied from + prefab objects. + +Examples: + +```bash +# Development: always create state +cmake -S . -B build-distillation -DAE_DISTILLATION=ON -DAE_FILTRATION=OFF + +# Hybrid operation: load existing state or create it +cmake -S . -B build-filtration -DAE_DISTILLATION=OFF -DAE_FILTRATION=ON + +# Production behavior: neither mode enabled +cmake -S . -B build-production -DAE_DISTILLATION=OFF -DAE_FILTRATION=OFF +``` + +`FS_INIT` may provide generated or static persisted-state maps. Persistence is +not implied by mutation; use the established application save path when state +must survive shutdown. + +### Formatting and Static Checks -`USER_CONFIG` selects the compile-time user-configuration header. The prescribed -operational default is `./config/user_config_hydrogen.h`. +Follow the repository's Google C++ style and warning policy. Keep includes +minimal and preserve intentional public umbrella includes with IWYU annotations. +Use the configured build's `compile_commands.json` for changed-file clang-tidy +checks. Regenerate the compilation database when compiler, CMake options, +platform, or user configuration changes. Apply formatting consistently with +the repository's existing `.clang-format` policy before submitting changes. For ESP-IDF, use the covered project at `projects/xtensa_lx6/vscode/aether-client-cpp`. Select the appropriate ESP32 From 9879c79e03d5648a953ba394b50a9e4e47cb333b Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Fri, 21 Aug 2026 11:43:25 +0500 Subject: [PATCH 2/3] remove false-positive bugrpone-assert checks --- .clang-tidy | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.clang-tidy b/.clang-tidy index c21f2118..db558504 100644 --- a/.clang-tidy +++ b/.clang-tidy @@ -39,10 +39,10 @@ Checks: - -cppcoreguidelines-pro-bounds-array-to-pointer-decay - -cppcoreguidelines-pro-type-static-cast-downcast - -bugprone-easily-swappable-parameters + - -bugprone-assert-side-effect - -readability-named-parameter - -readability-identifier-length CheckOptions: - bugprone-assert-side-effect.CheckFunctionCalls: true performance-move-const-arg.CheckTriviallyCopyableMove: false cppcoreguidelines-avoid-do-while.IgnoreMacros: true From 6648a96ea6bc100a027cab9e6c97e0d14a3169a5 Mon Sep 17 00:00:00 2001 From: BartolomeyKant Date: Fri, 21 Aug 2026 11:44:29 +0500 Subject: [PATCH 3/3] add save cloud server priority into persistent storage --- aether/ae_actions/ping.cpp | 9 +- aether/aether.cpp | 15 +- aether/client.cpp | 5 +- aether/cloud.cpp | 66 +++- aether/cloud.h | 22 +- aether/cloud_connections/cloud_request.cpp | 14 +- .../cloud_server_connection.cpp | 46 ++- .../cloud_server_connection.h | 25 +- .../cloud_server_connections.cpp | 243 ++++++++------- .../cloud_server_connections.h | 62 ++-- .../cloud_connections/ping_cloud_servers.cpp | 22 +- aether/cloud_connections/ping_cloud_servers.h | 4 +- .../client_cloud_manager.cpp | 32 +- .../connection_manager/client_cloud_manager.h | 8 +- .../connection_manager/get_cloud_aether.cpp | 3 +- .../root_server_select_stream.cpp | 21 +- .../registration/root_server_select_stream.h | 10 +- tests/test-server-connection/CMakeLists.txt | 2 + tests/test-server-connection/main.cpp | 10 + .../test_cloud_persistence.cpp | 293 ++++++++++++++++++ .../test_cloud_quarantine_loop.cpp | 149 ++++++++- .../test_registration_root_server_select.cpp | 134 ++++++++ 22 files changed, 952 insertions(+), 243 deletions(-) create mode 100644 tests/test-server-connection/test_cloud_persistence.cpp create mode 100644 tests/test-server-connection/test_registration_root_server_select.cpp diff --git a/aether/ae_actions/ping.cpp b/aether/ae_actions/ping.cpp index 02f21238..bef22008 100644 --- a/aether/ae_actions/ping.cpp +++ b/aether/ae_actions/ping.cpp @@ -67,7 +67,7 @@ Ping::Ping(AeContext const& ae_context, next_ping_hint_{next_ping_hint}, rx_window_{rx_window}, timeout_{timeout}, - server_id_{cloud_server_connection_->server()->server_id} { + server_id_{cloud_server_connection_->server_id()} { AE_TELE_INFO( kPing, "Ping action created to server id: {}, interval: {:%S}s, rx_window: " @@ -122,9 +122,10 @@ void Ping::Start(TimePoint current_time) { [this, req_id]() { PingResponseTimeout(req_id); }, current_time + timeout_); if (state_ == RequestState::kPending && !timeout_sub_) { - AE_TELE_ERROR(kPingTimeoutError, - "Ping timeout task allocation failed server id {} request {}", - server_id_, req_id); + AE_TELE_ERROR( + kPingTimeoutError, + "Ping timeout task allocation failed server id {} request {}", + server_id_, req_id); state_ = RequestState::kFinished; ResetRequestSubscriptions(); result_event_.Emit(PingResult{Error{5}}); diff --git a/aether/aether.cpp b/aether/aether.cpp index 63bde5d6..b75a866b 100644 --- a/aether/aether.cpp +++ b/aether/aether.cpp @@ -18,13 +18,13 @@ #include -#include "aether/obj/obj_ptr.h" #include "aether/client.h" -#include "aether/server.h" +#include "aether/obj/obj_ptr.h" #include "aether/registration_cloud.h" +#include "aether/server.h" -#include "aether/work_cloud.h" #include "aether/registration/registration.h" +#include "aether/work_cloud.h" #include "aether/aether_tele.h" @@ -75,9 +75,8 @@ Client::ptr Aether::CreateClient(ClientConfig const& config, auto client_cloud = WorkCloud::ptr::Create(domain, config.uid); [[maybe_unused]] auto res = // ~(^.^)~ - client_cloud.WithLoaded([&](auto const& cloud) { - cloud->SetServers(std::move(servers)); - }) && // ~(^.^)~ + client_cloud.WithLoaded( + [&](auto const& cloud) { cloud->SetServers(servers); }) && // ~(^.^)~ client.WithLoaded([&](auto const& client) { client->SetConfig(client_id, config.parent_uid, config.uid, config.ephemeral_uid, config.master_key, @@ -135,7 +134,9 @@ Client::ptr Aether::FindClient(std::string const& client_id) { void Aether::StoreClient(Client::ptr client) { assert(client.is_valid() && "Client is invalid"); - clients_[client.Load()->id()] = std::move(client); + auto const loaded_client = client.Load(); + assert(loaded_client && "Client failed to load"); + clients_[loaded_client->id()] = std::move(client); } SelectClientAction* Aether::FindSelectClientAction( diff --git a/aether/client.cpp b/aether/client.cpp index a3b79dec..fcd116f7 100644 --- a/aether/client.cpp +++ b/aether/client.cpp @@ -108,8 +108,9 @@ void Client::SetConfig(std::string client_id, Uid parent_uid, Uid uid, master_key_ = std::move(master_key); cloud_ = std::move(cloud); - for (auto& s : cloud_->servers()) { - server_keys_.emplace(s->server_id, ServerKeys{s->server_id, master_key_}); + for (auto const& server_entry : cloud_->servers()) { + auto const server_id = server_entry.first; + server_keys_.emplace(server_id, ServerKeys{server_id, master_key_}); } connectivity_policy_ = ClientConnectivityPolicy::ptr::Create( diff --git a/aether/cloud.cpp b/aether/cloud.cpp index f1977c06..b73997ec 100644 --- a/aether/cloud.cpp +++ b/aether/cloud.cpp @@ -16,25 +16,77 @@ #include "aether/cloud.h" +#include +#include +#include +#include +#include + namespace ae { Cloud::Cloud(ObjProp prop) : Obj{prop} {} void Cloud::AddServer(Server::ptr server) { - server.SetFlags(ObjFlags::kUnloadedByDefault); - servers_.emplace_back(std::move(server)); + [[maybe_unused]] auto const server_loaded = + server.WithLoaded([&](auto const& loaded_server) { + auto const server_id = loaded_server->server_id; + server.SetFlags(ObjFlags::kUnloadedByDefault); + auto it = servers_.find(server_id); + if (it != servers_.end()) { + it->second.server = std::move(server); + return; + } + + assert(servers_.size() < std::numeric_limits::max() && + "cloud server priority exceeds persisted capacity"); + auto priority = std::uint16_t{0}; + if (!servers_.empty()) { + auto const last_server = std::max_element( + servers_.begin(), servers_.end(), + [](auto const& left, auto const& right) { + return left.second.priority < right.second.priority; + }); + assert(last_server->second.priority < + std::numeric_limits::max() && + "cloud server priority exceeds persisted capacity"); + priority = + static_cast(last_server->second.priority + 1); + } + servers_.emplace(server_id, CloudServer{priority, std::move(server)}); + }); + assert(server_loaded && "cloud server must load"); cloud_updated_.Emit(); } -void Cloud::SetServers(std::vector servers) { - for (auto&& s : std::move(servers)) { - s.SetFlags(ObjFlags::kUnloadedByDefault); - servers_.emplace_back(std::move(s)); +void Cloud::SetServers(std::vector const& servers) { + assert(servers.size() < std::numeric_limits::max() && + "cloud server priority exceeds persisted capacity"); + servers_.clear(); + std::uint16_t priority = 0; + for (auto const& server : servers) { + auto stored_server = server; + stored_server.SetFlags(ObjFlags::kUnloadedByDefault); + auto const server_id = stored_server.WithLoaded( + [](auto const& loaded_server) { return loaded_server->server_id; }); + assert(server_id && "cloud server must load"); + if (!server_id) { + continue; + } + [[maybe_unused]] auto const inserted = + servers_ + .insert_or_assign(*server_id, + CloudServer{priority++, std::move(stored_server)}) + .second; + assert(inserted && "cloud server must not duplicate"); } cloud_updated_.Emit(); } -std::vector& Cloud::servers() { return servers_; } +std::map& Cloud::servers() { return servers_; } + +std::map const& Cloud::servers() const { + return servers_; +} EventSubscriber Cloud::cloud_updated() { return EventSubscriber{cloud_updated_}; diff --git a/aether/cloud.h b/aether/cloud.h index 2f5eeaf3..f5681751 100644 --- a/aether/cloud.h +++ b/aether/cloud.h @@ -17,14 +17,23 @@ #ifndef AETHER_CLOUD_H_ #define AETHER_CLOUD_H_ +#include +#include #include #include "aether/events/events.h" -#include "aether/server.h" #include "aether/obj/obj.h" +#include "aether/server.h" namespace ae { +struct CloudServer { + AE_REFLECT_MEMBERS(priority, server) + + std::uint16_t priority{}; + Server::ptr server; +}; + class Cloud : public Obj { AE_OBJECT(Cloud, Obj, 0) @@ -34,16 +43,19 @@ class Cloud : public Obj { public: explicit Cloud(ObjProp prop); - AE_OBJECT_REFLECT(AE_MMBRS(servers_)) + AE_OBJECT_REFLECT(AE_MMBR(servers_)) + // Requires no existing server priority is uint16_t's maximum; overflow is UB. void AddServer(Server::ptr server); - void SetServers(std::vector servers); + // Requires unique server IDs; duplicates are UB and debug-asserted. + void SetServers(std::vector const& servers); - std::vector& servers(); + std::map& servers(); + std::map const& servers() const; EventSubscriber cloud_updated(); private: - std::vector servers_; + std::map servers_; Event cloud_updated_; }; diff --git a/aether/cloud_connections/cloud_request.cpp b/aether/cloud_connections/cloud_request.cpp index 7ed6e4d4..0a972cc7 100644 --- a/aether/cloud_connections/cloud_request.cpp +++ b/aether/cloud_connections/cloud_request.cpp @@ -18,9 +18,9 @@ #include +#include "aether-miscpp/misc/override.h" #include "aether/aether.h" #include "aether/server.h" -#include "aether-miscpp/misc/override.h" #include "aether/write_action/write_action.h" #include "aether/cloud_connections/cloud_connections_tele.h" @@ -116,7 +116,7 @@ void CloudRequest::MakeRequest() { void CloudRequest::MakeServerRequest(CloudServerConnection* sc, ServerRequest& sr) { - AE_TELED_DEBUG("Make request to server {}", sc->server()->server_id); + AE_TELED_DEBUG("Make request to server {}", sc->server_id()); // Clear previous subscriptions and timeout sr.state_subs.Reset(); @@ -162,8 +162,7 @@ void CloudRequest::MakeServerRequest(CloudServerConnection* 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); + AE_TELED_WARNING("Request timeout for server {}", sc->server_id()); OnServerRequestTimeout(sc); }, request_timeout_); @@ -183,8 +182,7 @@ void CloudRequest::OnChannelChanged(CloudServerConnection* sc) { } auto& sr = it->second; if (sr.retry_count >= max_retries_) { - AE_TELED_WARNING("Server {} retry budget exhausted", - sc->server()->server_id); + AE_TELED_WARNING("Server {} retry budget exhausted", sc->server_id()); sr.exhausted = true; EnqueueMakeRequest(); return; @@ -202,7 +200,7 @@ void CloudRequest::OnWriteFailed(CloudServerConnection* sc) { sr.retry_count++; if (sr.retry_count >= max_retries_) { AE_TELED_WARNING("Server {} retry budget exhausted on write failure", - sc->server()->server_id); + sc->server_id()); sr.exhausted = true; } EnqueueMakeRequest(); @@ -216,7 +214,7 @@ void CloudRequest::OnServerRequestTimeout(CloudServerConnection* sc) { auto& sr = it->second; if (sr.retry_count >= max_retries_) { AE_TELED_WARNING("Server {} retry budget exhausted on timeout", - sc->server()->server_id); + sc->server_id()); sr.exhausted = true; EnqueueMakeRequest(); return; diff --git a/aether/cloud_connections/cloud_server_connection.cpp b/aether/cloud_connections/cloud_server_connection.cpp index 4cccca6c..aaaf578e 100644 --- a/aether/cloud_connections/cloud_server_connection.cpp +++ b/aether/cloud_connections/cloud_server_connection.cpp @@ -16,21 +16,20 @@ #include "aether/cloud_connections/cloud_server_connection.h" +#include +#include +#include + #include "aether/server.h" namespace ae { CloudServerConnection::CloudServerConnection( - Ptr const& server, IServerConnectionFactory& connection_factory) - : server_{server}, - connection_factory_{&connection_factory}, - priority_{}, - is_quarantined_{} {} - -std::size_t CloudServerConnection::priority() const { return priority_; } -void CloudServerConnection::SetPriority(std::size_t priority) { - priority_ = priority; -} + Ptr const& cloud, ServerId server_id, + IServerConnectionFactory& connection_factory) + : cloud_{cloud}, + server_id_{server_id}, + connection_factory_{&connection_factory} {} void CloudServerConnection::Restream() { if (client_connection_) { @@ -45,7 +44,7 @@ void CloudServerConnection::SetQuarantine(bool value) { bool CloudServerConnection::Connect() { client_connection_.reset(); - client_connection_ = connection_factory_->CreateConnection(server()); + client_connection_ = connection_factory_->CreateConnection(server().Load()); return static_cast(client_connection_); } @@ -58,10 +57,27 @@ ClientServerConnection* CloudServerConnection::client_connection() { return nullptr; } -Ptr CloudServerConnection::server() const { - auto server = server_.Lock(); - assert(server.get() != nullptr); - return server; +Server::ptr const& CloudServerConnection::server() const { + return cloud_server().server; +} + +std::size_t CloudServerConnection::priority() const { + return cloud_server().priority; +} + +void CloudServerConnection::SetPriority(std::size_t priority) { + assert(priority <= std::numeric_limits::max() && + "cloud server priority exceeds persisted capacity"); + cloud_server().priority = static_cast(priority); +} + +CloudServer& CloudServerConnection::cloud_server() const { + auto cloud = cloud_.Lock(); + assert(cloud && "cloud must outlive its connections"); + auto it = cloud->servers().find(server_id_); + assert(it != cloud->servers().end() && + "cloud server connection must have a persistent server entry"); + return it->second; } } // namespace ae diff --git a/aether/cloud_connections/cloud_server_connection.h b/aether/cloud_connections/cloud_server_connection.h index 1cfa4d18..3461b878 100644 --- a/aether/cloud_connections/cloud_server_connection.h +++ b/aether/cloud_connections/cloud_server_connection.h @@ -17,24 +17,22 @@ #ifndef AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTION_H_ #define AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTION_H_ +#include #include #include "aether/ptr/ptr_view.h" +#include "aether/cloud.h" +#include "aether/server.h" #include "aether/server_connections/client_server_connection.h" #include "aether/server_connections/iserver_connection_factory.h" namespace ae { -class Server; - class CloudServerConnection { public: - CloudServerConnection(Ptr const& server, + CloudServerConnection(Ptr const& cloud, ServerId server_id, IServerConnectionFactory& connection_factory); - std::size_t priority() const; - void SetPriority(std::size_t priority); - void Restream(); bool quarantine() const; @@ -45,14 +43,21 @@ class CloudServerConnection { ClientServerConnection* client_connection(); - Ptr server() const; + // The reference is valid only while the owning Cloud and server-map entry + // remain unchanged. + Server::ptr const& server() const; + ServerId server_id() const { return server_id_; } + std::size_t priority() const; + void SetPriority(std::size_t priority); private: - PtrView server_; + CloudServer& cloud_server() const; + + PtrView cloud_; + ServerId server_id_; IServerConnectionFactory* connection_factory_; std::shared_ptr client_connection_; - std::size_t priority_; - bool is_quarantined_; + bool is_quarantined_ = false; }; } // namespace ae diff --git a/aether/cloud_connections/cloud_server_connections.cpp b/aether/cloud_connections/cloud_server_connections.cpp index 7d7586dd..993eb898 100644 --- a/aether/cloud_connections/cloud_server_connections.cpp +++ b/aether/cloud_connections/cloud_server_connections.cpp @@ -18,33 +18,29 @@ #include #include #include -#include +#include #include #include "aether/api_protocol/api_protocol.h" -#include "aether/server.h" #include "aether/server_connections/server_connection.h" #include "aether/cloud_connections/cloud_connections_tele.h" // IWYU pragma: keep namespace ae { - -namespace { -auto SelectedServersLog( - [[maybe_unused]] std::vector const& - selected_servers) { #if DEBUG - std::vector server_ids; - server_ids.reserve(selected_servers.size()); - for (auto* server_connection : selected_servers) { - server_ids.emplace_back(server_connection->server()->server_id); +auto ServerList(std::vector const& servers) { + std::vector sids; + sids.reserve(servers.size()); + for (auto const* s : servers) { + sids.emplace_back(s->server_id()); } - return server_ids; + return sids; +} #else - return "!not a debug!"; -#endif +std::string_view ServerList(std::vector const&) { + return "!no debug!"; } -} // namespace +#endif static constexpr auto kCloudServerQuarantineTime = std::chrono::milliseconds{AE_CLOUD_SERVER_QUARANTINE_TIME_MS}; @@ -94,7 +90,9 @@ CloudServerConnections::CloudServerConnections( connection_factory_{std::move(connection_factory)}, max_connections_{max_connections} { InitServerConnections(); - ReconcileServers(); + if (max_connections_ != 0) { + ReconcileServers(); + } } CloudServerConnections::ServersUpdate::Subscriber @@ -130,85 +128,90 @@ void CloudServerConnections::Restream() { void CloudServerConnections::InitServerConnections() { auto cloud = cloud_.Lock(); - assert(cloud != nullptr); - server_connections_.clear(); - server_connections_.reserve(cloud->servers().size()); - for (auto& server : cloud->servers()) { - server_connections_.emplace_back(server.Load(), *connection_factory_); + assert(cloud && "cloud must outlive its connections"); + assert(server_entries_.empty() && + "server entries must be initialized only once"); + server_entries_.reserve(cloud->servers().size()); + for (auto const& cloud_server : cloud->servers()) { + server_entries_.emplace_back(cloud, cloud_server.first, + *connection_factory_); } - all_servers_.clear(); - all_servers_.reserve(server_connections_.size()); - for (auto& server : server_connections_) { - all_servers_.emplace_back(&server); + all_servers_.reserve(server_entries_.size()); + for (auto& entry : server_entries_) { + all_servers_.emplace_back(&entry.connection); } + std::sort(all_servers_.begin(), all_servers_.end(), + [](auto const* left, auto const* right) { + return left->priority() < right->priority(); + }); + NormalizeServerPriorities(); } -void CloudServerConnections::SubscribeToServerState( +bool CloudServerConnections::SubscribeToServerState( CloudServerConnection& server_connection) { auto* conn = server_connection.client_connection(); - if (conn == nullptr) { - return; + if (conn == nullptr || + conn->stream_info().link_state == LinkState::kLinkError) { + return false; } if (conn->stream_info().link_state == LinkState::kLinked) { AE_TELED_DEBUG("CLOUD_SERVER_LINKED server_id={} priority={}", - server_connection.server()->server_id, - server_connection.priority()); + server_connection.server_id(), server_connection.priority()); } - auto const key = reinterpret_cast(&server_connection); - auto& subs = server_subs_[key] = {}; - subs.state_sub = conn->stream_update_event().Subscribe( + auto& entry = ServerEntryFor(server_connection); + entry.state_sub = conn->stream_update_event().Subscribe( [this, sc{&server_connection}, conn]() { if (conn->stream_info().link_state == LinkState::kLinkError) { - QuarantineServer(*sc); + QuarantineAndReconcile(*sc); } else if (conn->stream_info().link_state == LinkState::kLinked) { AE_TELED_DEBUG("CLOUD_SERVER_LINKED server_id={} priority={}", - sc->server()->server_id, sc->priority()); + sc->server_id(), sc->priority()); } }); - subs.error_sub = conn->server_connection().server_error_event().Subscribe( - [this, sc{&server_connection}]() { QuarantineServer(*sc); }); + entry.error_sub = conn->server_connection().server_error_event().Subscribe( + [this, sc{&server_connection}]() { QuarantineAndReconcile(*sc); }); + return true; } void CloudServerConnections::UnsubscribeFromServerState( CloudServerConnection& server_connection) { - auto const key = reinterpret_cast(&server_connection); - auto it = server_subs_.find(key); - if (it == server_subs_.end()) { - return; - } - it->second.state_sub.Reset(); - it->second.error_sub.Reset(); - if (!it->second.quarantine_sub) { - server_subs_.erase(it); - } + auto& entry = ServerEntryFor(server_connection); + entry.state_sub.Reset(); + entry.error_sub.Reset(); } -void CloudServerConnections::QuarantineServer( +bool CloudServerConnections::QuarantineServer( CloudServerConnection& server_connection) { - auto const key = reinterpret_cast(&server_connection); if (server_connection.quarantine()) { - return; + return false; } AE_TELED_DEBUG("CLOUD_SERVER_QUARANTINED server_id={} priority={}", - server_connection.server()->server_id, - server_connection.priority()); + server_connection.server_id(), server_connection.priority()); UnsubscribeFromServerState(server_connection); - if (std::erase(selected_servers_, &server_connection) != 0) { - UpdateSelectedPriorities(); + auto const was_selected = + std::erase(selected_servers_, &server_connection) != 0; + assert(!server_entries_.empty() && + "quarantined server must belong to server connections"); + AE_TELED_DEBUG("CLOUD_SERVER_UNSELECTED server_id={}, selected list={}", + server_connection.server_id(), ServerList(selected_servers_)); + auto const server_it = + std::find(all_servers_.begin(), all_servers_.end(), &server_connection); + assert(server_it != all_servers_.end() && + "quarantined server must belong to all servers"); + std::rotate(server_it, server_it + 1, all_servers_.end()); + server_connection.SetQuarantine(true); + NormalizeServerPriorities(); + if (was_selected) { servers_update_event_.Emit(); } - server_connection.SetPriority(server_connections_.size()); - server_connection.SetQuarantine(true); server_quarantined_event_.Emit(&server_connection); // One delayed release: Disconnect + clear quarantine + reconcile. Do not // Disconnect on the error-callback stack. - auto& quarantine_sub = server_subs_[key].quarantine_sub; + auto& quarantine_sub = ServerEntryFor(server_connection).quarantine_sub; quarantine_sub = ae_context_.scheduler().DelayedTask( - [this, sc{&server_connection}, key]() { - ReleaseQuarantinedServer(*sc, key); - }, + [this, sc{&server_connection}]() { ReleaseQuarantinedServer(*sc); }, kCloudServerQuarantineTime); if (!quarantine_sub) { @@ -216,27 +219,38 @@ void CloudServerConnections::QuarantineServer( assert(false && "failed to schedule quarantine release"); } - ScheduleReconcileServers(); + return true; +} + +void CloudServerConnections::QuarantineAndReconcile( + CloudServerConnection& server_connection) { + if (QuarantineServer(server_connection)) { + ScheduleReconcileServers(); + } } void CloudServerConnections::ReleaseQuarantinedServer( - CloudServerConnection& server_connection, std::uintptr_t key) { + CloudServerConnection& server_connection) { if (!server_connection.quarantine()) { return; } AE_TELED_DEBUG("CLOUD_SERVER_RELEASED server_id={} priority={}", - server_connection.server()->server_id, - server_connection.priority()); - server_quarantine_release_event_.Emit(&server_connection); + server_connection.server_id(), server_connection.priority()); server_connection.Disconnect(); + auto const first_quarantined = + std::find_if(all_servers_.begin(), all_servers_.end(), + [](auto const* server) { return server->quarantine(); }); + auto const server_it = + std::find(all_servers_.begin(), all_servers_.end(), &server_connection); + assert(first_quarantined != all_servers_.end() && + "released server must be in quarantined suffix"); + assert(server_it != all_servers_.end() && + "released server must belong to all servers"); + std::rotate(first_quarantined, server_it, server_it + 1); server_connection.SetQuarantine(false); - auto it = server_subs_.find(key); - if (it != server_subs_.end()) { - it->second.quarantine_sub.Reset(); - if (!it->second.state_sub && !it->second.error_sub) { - server_subs_.erase(it); - } - } + NormalizeServerPriorities(); + server_quarantine_release_event_.Emit(&server_connection); + ServerEntryFor(server_connection).quarantine_sub.Reset(); ScheduleReconcileServers(); } @@ -261,75 +275,72 @@ void CloudServerConnections::ReconcileServers() { return; } // Vacancy-fill model: keep the current selected list stable and only append - // new non-selected candidates to fill available slots. - auto candidates = ReplacementCandidates(); - AE_TELED_DEBUG( - "Reconcile servers vacancy={} candidate_count={} selected_count={}", - max_connections_ - selected_servers_.size(), candidates.size(), - selected_servers_.size()); + // usable servers to fill available slots. The selected prefix is skipped; + // failed candidates move to the quarantined suffix, so retry the same index. + AE_TELED_DEBUG("Reconcile servers vacancy={} selected_count={}", + max_connections_ - selected_servers_.size(), + selected_servers_.size()); auto emplaced = false; - for (auto* candidate : candidates) { - if (selected_servers_.size() >= max_connections_) { - break; + auto candidate_index = selected_servers_.size(); + while (candidate_index < all_servers_.size() && + selected_servers_.size() < max_connections_) { + auto* candidate = all_servers_[candidate_index]; + if (candidate->quarantine()) { + ++candidate_index; + continue; } - candidate->SetPriority(selected_servers_.size()); - AE_TELED_DEBUG("CLOUD_SERVER_RECONNECT_ATTEMPT server_id={} priority={}", - candidate->server()->server_id, candidate->priority()); - candidate->Connect(); - auto* conn = candidate->client_connection(); - if (conn == nullptr || - conn->stream_info().link_state == LinkState::kLinkError) { + AE_TELED_DEBUG("CLOUD_SERVER_CONNECT_ATTEMPT server_id={} priority={}", + candidate->server_id(), candidate->priority()); + + if (!candidate->Connect() || !SubscribeToServerState(*candidate)) { AE_TELED_DEBUG( "Candidate unusable during reconcile server_id={} priority={}", - candidate->server()->server_id, candidate->priority()); - QuarantineServer(*candidate); + candidate->server_id(), candidate->priority()); + QuarantineAndReconcile(*candidate); continue; } + assert(candidate_index == selected_servers_.size() && + "candidate must follow selected prefix"); selected_servers_.emplace_back(candidate); - SubscribeToServerState(*candidate); + candidate_index = selected_servers_.size(); emplaced = true; } if (emplaced) { - AE_TELED_DEBUG("Selected servers reconciled selected_servers={}", - SelectedServersLog(selected_servers_)); + NormalizeServerPriorities(); + AE_TELED_DEBUG("Selected servers reconciled selected_count={}, list={}", + selected_servers_.size(), ServerList(selected_servers_)); servers_update_event_.Emit(); } } -bool CloudServerConnections::IsSelected(CloudServerConnection* sc) const { - return std::find(selected_servers_.begin(), selected_servers_.end(), sc) != - selected_servers_.end(); +CloudServerConnections::ServerEntry& CloudServerConnections::ServerEntryFor( + CloudServerConnection& server_connection) { + auto const entry_it = + std::find_if(server_entries_.begin(), server_entries_.end(), + [&server_connection](ServerEntry const& entry) { + return &entry.connection == &server_connection; + }); + assert(entry_it != server_entries_.end() && + "server connection must belong to server entries"); + return *entry_it; } -void CloudServerConnections::UpdateSelectedPriorities() { - for (std::size_t i = 0; i < selected_servers_.size(); ++i) { - selected_servers_.at(i)->SetPriority(i); +void CloudServerConnections::NormalizeServerPriorities() { + for (std::size_t i = 0; i < all_servers_.size(); ++i) { + all_servers_[i]->SetPriority(i); } } -auto CloudServerConnections::ReplacementCandidates() - -> std::vector { - std::vector servers; - servers.reserve(server_connections_.size()); - for (auto& s : server_connections_) { - if (!s.quarantine() && !IsSelected(&s)) { - servers.emplace_back(&s); - } - } - std::sort(servers.begin(), servers.end(), - [](auto const* left, auto const* right) { - return left->priority() < right->priority(); - }); - 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); + if (conn == nullptr) { + AE_TELED_WARNING("Skipping disconnected selected server"); + return; + } swas.emplace_back(&conn->AuthorizedApiCall( SubApi{[&](auto& api) { api_caller(api, sc); }})); }, diff --git a/aether/cloud_connections/cloud_server_connections.h b/aether/cloud_connections/cloud_server_connections.h index ca04a82b..7b7dc56f 100644 --- a/aether/cloud_connections/cloud_server_connections.h +++ b/aether/cloud_connections/cloud_server_connections.h @@ -16,9 +16,9 @@ #ifndef AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTIONS_H_ #define AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTIONS_H_ -#include #include #include +#include #include #include "aether/ae_context.h" @@ -56,6 +56,8 @@ class ReplicaWA final : public WriteAction { } // namespace cloud_server_connections_internal class CloudServerConnections { + friend struct CloudServerConnectionsTestAccess; + public: using ServersUpdate = Event; using ServerQuarantineEvent = Event; @@ -64,7 +66,6 @@ class 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. */ @@ -76,7 +77,11 @@ class CloudServerConnections { */ std::vector const& selected_servers() const; /** - * \brief List of all server connections including quarantined ones. + * \brief List of all server connections in canonical priority order. + * + * The list contains selected servers in selected_servers() order, followed + * by usable non-selected servers, then quarantined servers from oldest to + * newest quarantine. A server's persisted priority equals its index. */ std::vector const& servers(); @@ -122,62 +127,73 @@ class CloudServerConnections { return; } std::invoke(std::forward(func), - selected_servers_.at(priority.priority)); + selected_servers_[priority.priority]); } template void ForServersImpl(TFunc&& func, RequestPolicy::Replica replica) { + auto&& callable = std::forward(func); 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)); + std::invoke(callable, selected_servers_[i]); } } template void ForServersImpl(TFunc&& func, RequestPolicy::All) { + auto&& callable = std::forward(func); for (auto* sc : selected_servers_) { - std::invoke(std::forward(func), sc); + std::invoke(callable, sc); } } WriteAction& EmptyWriteAction(); WriteAction& ReplicaWriteAction(std::vector&& swas); + struct ServerEntry { + ServerEntry(Ptr const& cloud, ServerId server_id, + IServerConnectionFactory& connection_factory) + : connection{cloud, server_id, connection_factory} {} + + CloudServerConnection connection; + Subscription state_sub; + Subscription error_sub; + TaskSubscription quarantine_sub; + }; + void InitServerConnections(); // Caller must unsubscribe before re-subscribing the same server. - void SubscribeToServerState(CloudServerConnection& server_connection); + bool SubscribeToServerState(CloudServerConnection& server_connection); void UnsubscribeFromServerState(CloudServerConnection& server_connection); - void QuarantineServer(CloudServerConnection& server_connection); - void ReleaseQuarantinedServer(CloudServerConnection& server_connection, - std::uintptr_t key); + bool QuarantineServer(CloudServerConnection& server_connection); + void QuarantineAndReconcile(CloudServerConnection& server_connection); + void ReleaseQuarantinedServer(CloudServerConnection& server_connection); void ScheduleReconcileServers(); void ReconcileServers(); - bool IsSelected(CloudServerConnection* sc) const; - void UpdateSelectedPriorities(); - std::vector ReplacementCandidates(); - - struct ServerSubscriptions { - Subscription state_sub; - Subscription error_sub; - TaskSubscription quarantine_sub; - }; + void NormalizeServerPriorities(); + ServerEntry& ServerEntryFor(CloudServerConnection& server_connection); AeContext ae_context_; PtrView cloud_; std::unique_ptr connection_factory_; std::size_t max_connections_; - std::vector server_connections_; - + // Entries are reserved and populated once before connection pointers are + // published. Do not grow or rebuild this vector: callbacks retain pointers + // to its connections. + std::vector server_entries_; + + // Canonical priority order: selected prefix, usable non-selected servers, + // then quarantined servers oldest-to-newest. Persisted priorities equal + // all_servers_ indices. std::vector all_servers_; - // selected list of servers sorted by the priority + // Selected prefix of all_servers_, in selection and priority order. std::vector selected_servers_; ServersUpdate servers_update_event_; ServerQuarantineEvent server_quarantined_event_; ServerQuarantineEvent server_quarantine_release_event_; - std::map server_subs_; TaskSubscription defer_sub_; std::optional diff --git a/aether/cloud_connections/ping_cloud_servers.cpp b/aether/cloud_connections/ping_cloud_servers.cpp index 9aba9991..b1128e14 100644 --- a/aether/cloud_connections/ping_cloud_servers.cpp +++ b/aether/cloud_connections/ping_cloud_servers.cpp @@ -24,7 +24,6 @@ # include "aether/channels/channel.h" # include "aether/executors/executors.h" -# include "aether/server.h" # include "aether/cloud_connections/cloud_connections_tele.h" @@ -279,17 +278,16 @@ void PingCloudServers::DispatchToServers() { AE_TELED_ERROR("Visit empty cloud server connection!"); return; } - auto server = cloud_sc->server(); + auto const& server = cloud_sc->server(); if (server) { - ReconcileServer(server, *cloud_sc); + ReconcileServer(*cloud_sc); } }, policy_->rx_targets()); } -void PingCloudServers::ReconcileServer(Ptr const& server, - CloudServerConnection& cloud_sc) { - auto const server_id = server->server_id; +void PingCloudServers::ReconcileServer(CloudServerConnection& cloud_sc) { + auto const server_id = cloud_sc.server_id(); auto const priority = cloud_sc.priority(); auto it = server_pings_.find(server_id); @@ -309,11 +307,7 @@ void PingCloudServers::ServerQuarantined(CloudServerConnection* cloud_sc) { if (cloud_sc == nullptr) { return; } - auto server = cloud_sc->server(); - if (server == nullptr) { - return; - } - auto it = server_pings_.find(server->server_id); + auto it = server_pings_.find(cloud_sc->server_id()); if (it != server_pings_.end()) { it->second->Stop(); } @@ -324,11 +318,7 @@ void PingCloudServers::ServerQuarantineReleased( if (cloud_sc == nullptr) { return; } - auto server = cloud_sc->server(); - if (server == nullptr) { - return; - } - auto it = server_pings_.find(server->server_id); + auto it = server_pings_.find(cloud_sc->server_id()); if (it != server_pings_.end()) { server_pings_.erase(it); } diff --git a/aether/cloud_connections/ping_cloud_servers.h b/aether/cloud_connections/ping_cloud_servers.h index 215087e3..dd55ef65 100644 --- a/aether/cloud_connections/ping_cloud_servers.h +++ b/aether/cloud_connections/ping_cloud_servers.h @@ -33,7 +33,6 @@ # 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 { @@ -94,8 +93,7 @@ class PingCloudServers { private: void ServersUpdate(); void DispatchToServers(); - void ReconcileServer(Ptr const& server, - CloudServerConnection& cloud_sc); + void ReconcileServer(CloudServerConnection& cloud_sc); void ServerQuarantined(CloudServerConnection* cloud_sc); void ServerQuarantineReleased(CloudServerConnection* cloud_sc); diff --git a/aether/connection_manager/client_cloud_manager.cpp b/aether/connection_manager/client_cloud_manager.cpp index 22b14b54..5c1cd5ae 100644 --- a/aether/connection_manager/client_cloud_manager.cpp +++ b/aether/connection_manager/client_cloud_manager.cpp @@ -28,6 +28,8 @@ namespace ae { namespace client_cloud_manager_internal { +constexpr int kGetServersRequestError = 1; + GetCloudFromCache::GetCloudFromCache(AeContext const& ae_context, Cloud::ptr cloud) : cloud_{std::move(cloud)} { @@ -121,7 +123,7 @@ auto LoadMissing(Aether::ptr const& aether, Client::ptr const& client, BuildNewServers(aether, servers, res.value()); ex::set_value(std::move(ctx.receiver), std::move(servers)); } else { - ex::set_error(std::move(ctx.receiver), 1); + ex::set_error(std::move(ctx.receiver), kGetServersRequestError); } }); }); @@ -149,14 +151,18 @@ ClientCloudManager::ClientCloudManager(ObjProp prop, ObjPtr aether, ObjPtr client) : Obj{prop}, aether_{std::move(aether)}, client_{std::move(client)} { // save cloud cache for current client - Client::ptr{client}.WithLoaded([&](auto const& c) { - cloud_cache_.emplace(c->uid(), client_cloud_manager_internal::CloudCache{ - .version_confirmed = true, - .subject_uid = c->uid(), - .version = 0, - .cloud = c->cloud(), - }); - }); + [[maybe_unused]] auto const cache_initialized = + client_.WithLoaded([&](auto const& obj) { + auto* c = obj.template as(); + cloud_cache_.emplace(c->uid(), + client_cloud_manager_internal::CloudCache{ + .version_confirmed = true, + .subject_uid = c->uid(), + .version = 0, + .cloud = c->cloud(), + }); + }); + assert(cache_initialized && "Client did not load"); // init the rest Init(); @@ -253,7 +259,7 @@ void ClientCloudManager::CloudConfigs(std::vector const& configs) { FinalizeCloudConfig(conf); } else if (!it->second.finalizing && (it->second.version < conf.config_version)) { - it->second.version_confirmed = false, + it->second.version_confirmed = false; it->second.subject_uid = conf.subject_uid; it->second.version = conf.config_version; it->second.finalizing = true; @@ -294,8 +300,8 @@ void ClientCloudManager::FinalizeCloudConfig(CloudConfig const& conf) { })); } -Cloud::ptr ClientCloudManager::RegisterCloud(Uid uid, - std::vector servers) { +Cloud::ptr ClientCloudManager::RegisterCloud( + Uid uid, std::vector const& servers) { auto it = cloud_cache_.find(uid); assert((it != cloud_cache_.end()) && "Cloud should be in cache before register"); @@ -304,7 +310,7 @@ Cloud::ptr ClientCloudManager::RegisterCloud(Uid uid, if (!it->second.cloud.is_valid()) { it->second.cloud = WorkCloud::ptr::Create(domain, uid); } - it->second.cloud.Load()->SetServers(std::move(servers)); + it->second.cloud.Load()->SetServers(servers); return it->second.cloud; } diff --git a/aether/connection_manager/client_cloud_manager.h b/aether/connection_manager/client_cloud_manager.h index c1d719cf..571fd182 100644 --- a/aether/connection_manager/client_cloud_manager.h +++ b/aether/connection_manager/client_cloud_manager.h @@ -20,13 +20,13 @@ #include #include +#include "aether/actions/action_pool.h" #include "aether/cloud.h" +#include "aether/events/events.h" +#include "aether/executors/executors.h" #include "aether/obj/obj.h" #include "aether/ptr/ptr.h" #include "aether/types/uid.h" -#include "aether/events/events.h" -#include "aether/actions/action_pool.h" -#include "aether/executors/executors.h" #include "aether/ae_actions/get_servers.h" #include "aether/cloud_connections/cloud_subscription.h" @@ -104,7 +104,7 @@ class ClientCloudManager : public Obj { void CloudConfigs(std::vector const& configs); void FinalizeCloudConfig(CloudConfig const& conf); auto MakeServersSender(std::vector const& sids); - Cloud::ptr RegisterCloud(Uid uid, std::vector servers); + Cloud::ptr RegisterCloud(Uid uid, std::vector const& servers); GetCloudActionPool& get_cloud_action_pool(); diff --git a/aether/connection_manager/get_cloud_aether.cpp b/aether/connection_manager/get_cloud_aether.cpp index 81065442..1ab88a25 100644 --- a/aether/connection_manager/get_cloud_aether.cpp +++ b/aether/connection_manager/get_cloud_aether.cpp @@ -34,8 +34,7 @@ GetCloudFromAether::GetCloudFromAether(AeContext const& ae_context, 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); + client_uid_, server_connection->server_id()); auth_api->report_applied_config(std::vector{AppliedConfig{ .subject_uid = client_uid_, .config_version = -1, diff --git a/aether/registration/root_server_select_stream.cpp b/aether/registration/root_server_select_stream.cpp index 0e4daed5..6b520b3d 100644 --- a/aether/registration/root_server_select_stream.cpp +++ b/aether/registration/root_server_select_stream.cpp @@ -26,8 +26,7 @@ RootServerSelectStream::RootServerSelectStream( : ae_context_{ae_context}, cloud_{cloud}, buffer_write_{ae_context, - MethodPtr<&RootServerSelectStream::OnWrite>{this}}, - server_index_{} { + MethodPtr<&RootServerSelectStream::OnWrite>{this}} { SelectServer(); } @@ -84,13 +83,25 @@ void RootServerSelectStream::SelectServer() { auto cloud_ptr = cloud_.Lock(); assert(cloud_ptr); - if (server_index_ >= cloud_ptr->servers().size()) { + auto const& servers = cloud_ptr->servers(); + auto server_it = servers.end(); + // Registration server priorities must be contiguous starting at zero; + // gaps and equal-priority registration candidates are UB. + for (auto it = servers.begin(); it != servers.end(); ++it) { + auto const priority = it->second.priority; + if (priority == server_priority_) { + server_it = it; + break; + } + } + if (server_it == servers.end()) { CloudError(); return; } - auto& chosen_server = cloud_ptr->servers()[server_index_++]; + ++server_priority_; + auto const& chosen_server = server_it->second; - server_connection_.emplace(ae_context_, chosen_server.Load()); + server_connection_.emplace(ae_context_, chosen_server.server.Load()); server_connection_->out_data_event().Subscribe(out_data_event_); server_connection_->server_error_event().Subscribe( diff --git a/aether/registration/root_server_select_stream.h b/aether/registration/root_server_select_stream.h index 5af78682..2a14bd75 100644 --- a/aether/registration/root_server_select_stream.h +++ b/aether/registration/root_server_select_stream.h @@ -21,6 +21,7 @@ #if AE_SUPPORT_REGISTRATION +# include # include # include "aether/ae_context.h" @@ -33,6 +34,7 @@ namespace ae { class Aether; +struct RootServerSelectStreamTestAccess; class RootServerSelectStream final : public ByteIStream { public: static constexpr std::size_t kBufferCapacity = 200; @@ -52,6 +54,8 @@ class RootServerSelectStream final : public ByteIStream { CloudErrorEvent::Subscriber cloud_error_event(); private: + friend struct RootServerSelectStreamTestAccess; + WriteAction* OnWrite(DataBuffer&& data); void SelectServer(); @@ -62,7 +66,11 @@ class RootServerSelectStream final : public ByteIStream { PtrView cloud_; BufferWrite buffer_write_; - std::size_t server_index_; + // The next server priority to select. Registration server priorities must be + // contiguous starting at zero; gaps are UB. This invariant and Cloud's + // strict server collection capacity bound ensure this value never increments + // past the representable range. Overflow is unsupported and UB. + std::uint16_t server_priority_{0}; std::optional server_connection_; StreamUpdateEvent stream_update_event_; diff --git a/tests/test-server-connection/CMakeLists.txt b/tests/test-server-connection/CMakeLists.txt index 121b6c35..d1b6cc18 100644 --- a/tests/test-server-connection/CMakeLists.txt +++ b/tests/test-server-connection/CMakeLists.txt @@ -18,6 +18,8 @@ list(APPEND test_srcs main.cpp test_server_connection_recovery.cpp test_cloud_quarantine_loop.cpp + test_cloud_persistence.cpp + test_registration_root_server_select.cpp ${CMAKE_CURRENT_LIST_DIR}/../test-object-system/map_domain_storage.cpp ) diff --git a/tests/test-server-connection/main.cpp b/tests/test-server-connection/main.cpp index f190391e..9214d9da 100644 --- a/tests/test-server-connection/main.cpp +++ b/tests/test-server-connection/main.cpp @@ -16,15 +16,25 @@ #include +#include "aether/config.h" + void setUp() {} void tearDown() {} extern int run_test_server_connection_recovery(); extern int run_test_cloud_quarantine_loop(); +extern int run_test_cloud_persistence(); +#if AE_SUPPORT_REGISTRATION +extern int run_test_registration_root_server_select(); +#endif // AE_SUPPORT_REGISTRATION int main() { int res = 0; res += run_test_server_connection_recovery(); res += run_test_cloud_quarantine_loop(); + res += run_test_cloud_persistence(); +#if AE_SUPPORT_REGISTRATION + res += run_test_registration_root_server_select(); +#endif // AE_SUPPORT_REGISTRATION return res; } diff --git a/tests/test-server-connection/test_cloud_persistence.cpp b/tests/test-server-connection/test_cloud_persistence.cpp new file mode 100644 index 00000000..8dd7c0ef --- /dev/null +++ b/tests/test-server-connection/test_cloud_persistence.cpp @@ -0,0 +1,293 @@ +/* + * 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 "aether/adapter_registry.h" +#include "aether/ae_context.h" +#include "aether/cloud.h" +#include "aether/cloud_connections/cloud_server_connections.h" +#include "aether/obj/domain.h" +#include "aether/server.h" +#include "aether/server_connections/client_server_connection.h" +#include "aether/server_connections/iserver_connection_factory.h" +#include "aether/types/address.h" +#include "aether/types/server_id.h" +#include "aether/work_cloud.h" + +#include "tests/test-object-system/map_domain_storage.h" + +namespace ae { +namespace test_cloud_persistence { +struct TestContext { + AeCtx ToAeContext() const { + static constexpr auto table = + AeCtxTable{nullptr, [](void* obj) -> TaskScheduler& { + return static_cast(obj)->sched; + }}; + return AeCtx{const_cast(this), &table}; // NOLINT + } + + TaskScheduler sched; +}; + +class CountingNullFactory final : public IServerConnectionFactory { + public: + std::shared_ptr CreateConnection( + Ptr const& /*server*/) override { + ++attempts; + return {}; + } + int attempts{0}; +}; + +struct CloudFixture { + CloudFixture(std::unique_ptr factory, + IServerConnectionFactory* raw) + : ae_ctx{ctx}, + domain{Now(), storage}, + registry{AdapterRegistry::ptr::Create(CreateWith{domain})}, + server{Server::ptr::Create(CreateWith{domain}, ServerId{7}, + std::vector{}, registry)}, + cloud{Cloud::ptr::Create(CreateWith{domain})}, + factory_raw{raw} { + cloud->AddServer(server); + connections = std::make_unique( + ae_ctx, cloud.Load(), std::move(factory), /*max*/ 1); + } + + TestContext ctx; + AeContext ae_ctx; + MapDomainStorage storage; + Domain domain; + AdapterRegistry::ptr registry; + Server::ptr server; + Cloud::ptr cloud; + IServerConnectionFactory* factory_raw{nullptr}; + std::unique_ptr connections; +}; + +void test_CloudSetServersReplacesEntries() { + MapDomainStorage storage; + Domain domain{Now(), storage}; + auto registry = AdapterRegistry::ptr::Create(CreateWith{domain}); + auto first = Server::ptr::Create(CreateWith{domain}, ServerId{7}, + std::vector{}, registry); + auto second = Server::ptr::Create(CreateWith{domain}, ServerId{8}, + std::vector{}, registry); + auto third = Server::ptr::Create(CreateWith{domain}, ServerId{9}, + std::vector{}, registry); + auto fourth = Server::ptr::Create(CreateWith{domain}, ServerId{10}, + std::vector{}, registry); + auto cloud = Cloud::ptr::Create(CreateWith{domain}); + + cloud->SetServers({first, second}); + cloud->SetServers({third, fourth}); + + TEST_ASSERT_EQUAL_UINT(2, cloud->servers().size()); + TEST_ASSERT_FALSE(cloud->servers().contains(ServerId{7})); + TEST_ASSERT_FALSE(cloud->servers().contains(ServerId{8})); + TEST_ASSERT_EQUAL_UINT(0, cloud->servers().at(ServerId{9}).priority); + TEST_ASSERT_EQUAL_UINT(1, cloud->servers().at(ServerId{10}).priority); +} + +void test_CloudAddServerAppendsAndPreservesExistingPriority() { + auto factory = std::make_unique(); + auto* factory_raw = factory.get(); + CloudFixture f{std::move(factory), factory_raw}; + f.connections.reset(); + auto second_server = Server::ptr::Create(CreateWith{f.domain}, ServerId{8}, + std::vector{}, f.registry); + auto replacement = Server::ptr::Create(CreateWith{f.domain}, ServerId{7}, + std::vector{}, f.registry); + + f.cloud->servers().at(ServerId{7}).priority = 12; + f.cloud->AddServer(second_server); + f.cloud->AddServer(replacement); + + TEST_ASSERT_EQUAL_UINT(13, f.cloud->servers().at(ServerId{8}).priority); + TEST_ASSERT_EQUAL_UINT(12, f.cloud->servers().at(ServerId{7}).priority); + TEST_ASSERT_EQUAL_UINT(replacement.id().id(), + f.cloud->servers().at(ServerId{7}).server.id().id()); +} + +void test_CloudServerConnectionPriorityUsesCloudMap() { + auto factory = std::make_unique(); + auto* factory_raw = factory.get(); + CloudFixture f{std::move(factory), factory_raw}; + + auto* server_connection = f.connections->servers().front(); + TEST_ASSERT_EQUAL_UINT(0, server_connection->priority()); + server_connection->SetPriority(12); + TEST_ASSERT_EQUAL_UINT(12, server_connection->priority()); + TEST_ASSERT_EQUAL_UINT(12, f.cloud->servers().at(ServerId{7}).priority); +} + +void test_CloudEqualPrioritiesDoNotRequireTieOrder() { + auto factory = std::make_unique(); + auto* factory_raw = factory.get(); + CloudFixture f{std::move(factory), factory_raw}; + f.connections.reset(); + auto second = Server::ptr::Create(CreateWith{f.domain}, ServerId{8}, + std::vector{}, f.registry); + f.cloud->AddServer(second); + f.cloud->servers().at(ServerId{7}).priority = 0; + f.cloud->servers().at(ServerId{8}).priority = 0; + + auto reordered_factory = std::make_unique(); + auto* reordered_factory_raw = reordered_factory.get(); + CloudServerConnections connections{f.ae_ctx, f.cloud.Load(), + std::move(reordered_factory), /*max*/ 0}; + + auto const& servers = connections.servers(); + TEST_ASSERT_EQUAL_UINT(2, servers.size()); + TEST_ASSERT_TRUE((servers[0]->server_id() == ServerId{7} && + servers[1]->server_id() == ServerId{8}) || + (servers[0]->server_id() == ServerId{8} && + servers[1]->server_id() == ServerId{7})); + TEST_ASSERT_EQUAL_UINT(0, servers[0]->priority()); + TEST_ASSERT_EQUAL_UINT(1, servers[1]->priority()); + TEST_ASSERT_EQUAL_INT(0, reordered_factory_raw->attempts); +} + +void test_CloudServerConnectionServerReferencesCloudMapEntry() { + auto factory = std::make_unique(); + auto* factory_raw = factory.get(); + CloudFixture f{std::move(factory), factory_raw}; + + auto* server_connection = f.connections->servers().front(); + auto const& server = server_connection->server(); + + TEST_ASSERT_EQUAL_PTR(&f.cloud->servers().at(ServerId{7}).server, &server); + TEST_ASSERT_TRUE(server.is_loaded()); + TEST_ASSERT_EQUAL_UINT(7, server->server_id); +} + +void test_CloudServerConnectionPriorityRoundTripsAndRestoresSelectionOrder() { + MapDomainStorage storage; + Domain domain{Now(), storage}; + auto registry = AdapterRegistry::ptr::Create(CreateWith{domain}); + auto first = Server::ptr::Create(CreateWith{domain}, ServerId{10}, + std::vector{}, registry); + auto second = Server::ptr::Create(CreateWith{domain}, ServerId{20}, + std::vector{}, registry); + auto third = Server::ptr::Create(CreateWith{domain}, ServerId{30}, + std::vector{}, registry); + auto cloud = WorkCloud::ptr::Create(CreateWith{domain}.with_id(100), Uid{}); + cloud->AddServer(first); + cloud->AddServer(second); + cloud->AddServer(third); + int initial_attempts{}; + std::size_t initial_overflows{}; + { + TestContext initial_context; + AeContext initial_ae_context{initial_context}; + auto initial_factory = std::make_unique(); + auto* initial_factory_raw = initial_factory.get(); + CloudServerConnections initial_connections{initial_ae_context, cloud.Load(), + std::move(initial_factory), + /*max*/ 0}; + + initial_connections.servers().at(0)->SetPriority(2); + initial_connections.servers().at(1)->SetPriority(1); + initial_connections.servers().at(2)->SetPriority(0); + initial_attempts = initial_factory_raw->attempts; + initial_overflows = initial_context.sched.overflow_counter(); + cloud.Save(); + } + TEST_ASSERT_EQUAL_INT(0, initial_attempts); + TEST_ASSERT_EQUAL_UINT(0, initial_overflows); + + Domain restarted_domain{Now(), storage}; + auto restored_cloud = + WorkCloud::ptr::Declare(CreateWith{restarted_domain}.with_id(cloud.id())); + restored_cloud.Load(); + + TEST_ASSERT_EQUAL_UINT(3, restored_cloud->servers().size()); + TEST_ASSERT_EQUAL_UINT(2, + restored_cloud->servers().at(ServerId{10}).priority); + TEST_ASSERT_EQUAL_UINT(1, + restored_cloud->servers().at(ServerId{20}).priority); + TEST_ASSERT_EQUAL_UINT(0, + restored_cloud->servers().at(ServerId{30}).priority); + + struct RestoredConnectionState { + int attempts{}; + std::size_t overflows{}; + std::vector server_ids; + } restored_connection_state; + { + TestContext restored_context; + AeContext restored_ae_context{restored_context}; + auto factory = std::make_unique(); + auto* factory_raw = factory.get(); + CloudServerConnections connections{restored_ae_context, + restored_cloud.Load(), + std::move(factory), /*max*/ 0}; + + auto const& servers = connections.servers(); + restored_connection_state.attempts = factory_raw->attempts; + restored_connection_state.overflows = + restored_context.sched.overflow_counter(); + for (std::size_t i = 0; i < servers.size(); ++i) { + restored_connection_state.server_ids.emplace_back( + servers[i]->server_id()); + } + } + TEST_ASSERT_EQUAL_INT(0, restored_connection_state.attempts); + TEST_ASSERT_EQUAL_UINT(0, restored_connection_state.overflows); + TEST_ASSERT_EQUAL_UINT(2, + restored_cloud->servers().at(ServerId{10}).priority); + TEST_ASSERT_EQUAL_UINT(1, + restored_cloud->servers().at(ServerId{20}).priority); + TEST_ASSERT_EQUAL_UINT(0, + restored_cloud->servers().at(ServerId{30}).priority); + TEST_ASSERT_EQUAL_UINT(ServerId{30}, + restored_connection_state.server_ids.at(0)); + TEST_ASSERT_EQUAL_UINT(ServerId{20}, + restored_connection_state.server_ids.at(1)); + TEST_ASSERT_EQUAL_UINT(ServerId{10}, + restored_connection_state.server_ids.at(2)); + for (std::size_t i = 0; i < restored_connection_state.server_ids.size(); + ++i) { + TEST_ASSERT_EQUAL_UINT(static_cast(i), + static_cast( + restored_cloud->servers() + .at(restored_connection_state.server_ids[i]) + .priority)); + } +} + +} // namespace test_cloud_persistence +} // namespace ae + +int run_test_cloud_persistence() { + using namespace ae::test_cloud_persistence; // NOLINT + + UNITY_BEGIN(); + RUN_TEST(test_CloudSetServersReplacesEntries); + RUN_TEST(test_CloudAddServerAppendsAndPreservesExistingPriority); + RUN_TEST(test_CloudServerConnectionPriorityUsesCloudMap); + RUN_TEST(test_CloudEqualPrioritiesDoNotRequireTieOrder); + RUN_TEST(test_CloudServerConnectionServerReferencesCloudMapEntry); + RUN_TEST( + test_CloudServerConnectionPriorityRoundTripsAndRestoresSelectionOrder); + return UNITY_END(); +} diff --git a/tests/test-server-connection/test_cloud_quarantine_loop.cpp b/tests/test-server-connection/test_cloud_quarantine_loop.cpp index 7622df65..8179de30 100644 --- a/tests/test-server-connection/test_cloud_quarantine_loop.cpp +++ b/tests/test-server-connection/test_cloud_quarantine_loop.cpp @@ -14,7 +14,9 @@ * limitations under the License. */ +#include #include +#include #include #include #include @@ -32,10 +34,38 @@ #include "aether/server_connections/iserver_connection_factory.h" #include "aether/types/address.h" #include "aether/types/server_id.h" +#include "aether/work_cloud.h" #include "tests/test-object-system/map_domain_storage.h" namespace ae { +struct CloudServerConnectionsTestAccess { + static bool Quarantine(CloudServerConnections& connections, + CloudServerConnection& server) { + return connections.QuarantineServer(server); + } + + static void Release(CloudServerConnections& connections, + CloudServerConnection& server) { + connections.ReleaseQuarantinedServer(server); + } + + static bool HasStateSubscription(CloudServerConnections& connections, + CloudServerConnection& server) { + return static_cast(connections.ServerEntryFor(server).state_sub); + } + + static bool HasErrorSubscription(CloudServerConnections& connections, + CloudServerConnection& server) { + return static_cast(connections.ServerEntryFor(server).error_sub); + } + + static bool HasQuarantineTask(CloudServerConnections& connections, + CloudServerConnection& server) { + return static_cast(connections.ServerEntryFor(server).quarantine_sub); + } +}; + namespace test_cloud_quarantine_loop { struct TestContext { AeCtx ToAeContext() const { @@ -78,7 +108,9 @@ class SwitchableNullFactory final : public IServerConnectionFactory { struct CloudFixture { CloudFixture(std::unique_ptr factory, - IServerConnectionFactory* raw) + IServerConnectionFactory* raw, + std::vector additional_server_ids = {}, + std::size_t max_connections = 1) : ae_ctx{ctx}, domain{Now(), storage}, registry{AdapterRegistry::ptr::Create(CreateWith{domain})}, @@ -87,8 +119,12 @@ struct CloudFixture { cloud{Cloud::ptr::Create(CreateWith{domain})}, factory_raw{raw} { cloud->AddServer(server); + for (auto server_id : additional_server_ids) { + cloud->AddServer(Server::ptr::Create(CreateWith{domain}, server_id, + std::vector{}, registry)); + } connections = std::make_unique( - ae_ctx, cloud.Load(), std::move(factory), /*max*/ 1); + ae_ctx, cloud.Load(), std::move(factory), max_connections); } bool AnyQuarantined() const { @@ -111,6 +147,47 @@ struct CloudFixture { std::unique_ptr connections; }; +void AssertCanonicalPriorities(CloudServerConnections& connections) { + auto const& servers = connections.servers(); + for (std::size_t i = 0; i < servers.size(); ++i) { + TEST_ASSERT_EQUAL_UINT(static_cast(i), + static_cast(servers[i]->priority())); + } +} + +void test_CloudQuarantineAndReleasePreserveCanonicalOrder() { + auto factory = std::make_unique(); + auto* factory_raw = factory.get(); + CloudFixture f{ + std::move(factory), factory_raw, {ServerId{8}, ServerId{9}}, /*max*/ 0}; + auto& servers = f.connections->servers(); + + TEST_ASSERT_EQUAL_UINT(ServerId{7}, servers[0]->server_id()); + TEST_ASSERT_EQUAL_UINT(ServerId{8}, servers[1]->server_id()); + TEST_ASSERT_EQUAL_UINT(ServerId{9}, servers[2]->server_id()); + AssertCanonicalPriorities(*f.connections); + + TEST_ASSERT_TRUE(CloudServerConnectionsTestAccess::Quarantine(*f.connections, + *servers[0])); + TEST_ASSERT_EQUAL_UINT(ServerId{8}, servers[0]->server_id()); + TEST_ASSERT_EQUAL_UINT(ServerId{9}, servers[1]->server_id()); + TEST_ASSERT_EQUAL_UINT(ServerId{7}, servers[2]->server_id()); + AssertCanonicalPriorities(*f.connections); + + TEST_ASSERT_TRUE(CloudServerConnectionsTestAccess::Quarantine(*f.connections, + *servers[0])); + TEST_ASSERT_EQUAL_UINT(ServerId{9}, servers[0]->server_id()); + TEST_ASSERT_EQUAL_UINT(ServerId{7}, servers[1]->server_id()); + TEST_ASSERT_EQUAL_UINT(ServerId{8}, servers[2]->server_id()); + AssertCanonicalPriorities(*f.connections); + + CloudServerConnectionsTestAccess::Release(*f.connections, *servers[1]); + TEST_ASSERT_EQUAL_UINT(ServerId{9}, servers[0]->server_id()); + TEST_ASSERT_EQUAL_UINT(ServerId{7}, servers[1]->server_id()); + TEST_ASSERT_EQUAL_UINT(ServerId{8}, servers[2]->server_id()); + AssertCanonicalPriorities(*f.connections); +} + void test_CloudQuarantineDoesNotBusyLoop() { auto factory = std::make_unique(); auto* factory_raw = factory.get(); @@ -122,6 +199,7 @@ void test_CloudQuarantineDoesNotBusyLoop() { "expected first attempt"); TEST_ASSERT_TRUE_MESSAGE(f.AnyQuarantined(), "expected quarantine"); TEST_ASSERT_EQUAL_UINT(0, f.connections->count_connections()); + TEST_ASSERT_EQUAL_UINT(0, f.cloud->servers().at(ServerId{7}).priority); auto attempts_after_first = factory_raw->attempts; f.ctx.PumpAt(t0, 128); @@ -133,6 +211,70 @@ void test_CloudQuarantineDoesNotBusyLoop() { TEST_ASSERT_EQUAL_INT(attempts_after_first, factory_raw->attempts); } +void test_CloudImmediateUnusableCandidatesUseQuarantinePath() { + auto factory = std::make_unique(); + auto* factory_raw = factory.get(); + CloudFixture f{std::move(factory), factory_raw, {ServerId{8}}, /*max*/ 2}; + TEST_ASSERT_TRUE(f.AnyQuarantined()); + + std::vector quarantined_server_ids; + auto quarantined_sub = f.connections->server_quarantined_event().Subscribe( + [&](CloudServerConnection* server) { + quarantined_server_ids.emplace_back(server->server_id()); + }); + + // Construction performs the first reconciliation before the event + // subscription. Retry both immediately unusable candidates after quarantine + // release so their SubscribeToServerState failures are observable here. + auto const retry_at = + std::chrono::system_clock::now() + + std::chrono::milliseconds{AE_CLOUD_SERVER_QUARANTINE_TIME_MS + 1}; + // Release the original quarantine tasks at the simulated expiry. Run the + // resulting reconciliation at wall-clock time so quarantine tasks created + // by the immediately unusable candidates are not already due. + f.ctx.PumpAt(retry_at, 1); + f.ctx.PumpAt(std::chrono::system_clock::now(), 1); + + TEST_ASSERT_TRUE_MESSAGE(quarantined_server_ids.size() >= 2, + "expected both candidates to be quarantined"); + TEST_ASSERT_TRUE(std::find(quarantined_server_ids.begin(), + quarantined_server_ids.end(), + ServerId{7}) != quarantined_server_ids.end()); + TEST_ASSERT_TRUE(std::find(quarantined_server_ids.begin(), + quarantined_server_ids.end(), + ServerId{8}) != quarantined_server_ids.end()); + TEST_ASSERT_EQUAL_UINT(0, f.connections->count_connections()); + for (auto* server : f.connections->servers()) { + TEST_ASSERT_TRUE(server->quarantine()); + TEST_ASSERT_FALSE(CloudServerConnectionsTestAccess::HasStateSubscription( + *f.connections, *server)); + TEST_ASSERT_FALSE(CloudServerConnectionsTestAccess::HasErrorSubscription( + *f.connections, *server)); + TEST_ASSERT_TRUE(CloudServerConnectionsTestAccess::HasQuarantineTask( + *f.connections, *server)); + } +} + +void test_CloudServerPointersRemainStableAcrossQuarantine() { + auto factory = std::make_unique(); + auto* factory_raw = factory.get(); + CloudFixture f{ + std::move(factory), factory_raw, {ServerId{8}, ServerId{9}}, /*max*/ 0}; + auto const pointers = f.connections->servers(); + + TEST_ASSERT_TRUE(CloudServerConnectionsTestAccess::Quarantine(*f.connections, + *pointers[0])); + CloudServerConnectionsTestAccess::Release(*f.connections, *pointers[0]); + TEST_ASSERT_FALSE(CloudServerConnectionsTestAccess::HasQuarantineTask( + *f.connections, *pointers[0])); + + for (auto* pointer : pointers) { + auto const current = std::find(f.connections->servers().begin(), + f.connections->servers().end(), pointer); + TEST_ASSERT_TRUE(current != f.connections->servers().end()); + } +} + void test_CloudQuarantineReleaseAfterExpiry() { auto factory = std::make_unique(); auto* factory_raw = factory.get(); @@ -224,7 +366,10 @@ int run_test_cloud_quarantine_loop() { using namespace ae::test_cloud_quarantine_loop; // NOLINT UNITY_BEGIN(); + RUN_TEST(test_CloudQuarantineAndReleasePreserveCanonicalOrder); RUN_TEST(test_CloudQuarantineDoesNotBusyLoop); + RUN_TEST(test_CloudImmediateUnusableCandidatesUseQuarantinePath); + RUN_TEST(test_CloudServerPointersRemainStableAcrossQuarantine); RUN_TEST(test_CloudQuarantineReleaseAfterExpiry); RUN_TEST(test_CloudQuarantineNoRecursiveLoopSameTimestamp); RUN_TEST(test_CloudQuarantineAttemptsBoundedOverSimulatedSecond); diff --git a/tests/test-server-connection/test_registration_root_server_select.cpp b/tests/test-server-connection/test_registration_root_server_select.cpp new file mode 100644 index 00000000..0278076b --- /dev/null +++ b/tests/test-server-connection/test_registration_root_server_select.cpp @@ -0,0 +1,134 @@ +/* + * 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 "aether/config.h" + +#if AE_SUPPORT_REGISTRATION + +# include "aether/adapter_registry.h" +# include "aether/ae_context.h" +# include "aether/aether.h" +# include "aether/obj/domain.h" +# include "aether/registration/root_server_select_stream.h" +# include "aether/registration_cloud.h" +# include "aether/server.h" +# include "aether/types/address.h" +# include "aether/types/server_id.h" + +# include "tests/test-object-system/map_domain_storage.h" + +namespace ae { + +struct RootServerSelectStreamTestAccess { + static std::uint16_t ServerPriority(RootServerSelectStream const& stream) { + return stream.server_priority_; + } + + static void ServerError(RootServerSelectStream& stream) { + stream.ServerError(); + } +}; + +namespace test_registration_root_server_select { + +struct TestContext { + AeCtx ToAeContext() const { + static constexpr auto table = + AeCtxTable{nullptr, [](void* obj) -> TaskScheduler& { + return static_cast(obj)->scheduler; + }}; + return AeCtx{const_cast(this), &table}; // NOLINT + } + + TaskScheduler scheduler; +}; + +struct Fixture { + Fixture() + : ae_context{context}, + domain{Now(), storage}, + aether{Aether::ptr::Create(CreateWith{domain})}, + registry{AdapterRegistry::ptr::Create(CreateWith{domain})}, + cloud{RegistrationCloud::ptr::Create(CreateWith{domain}, aether)} {} + + Server::ptr AddServer(ServerId id, std::uint16_t priority) { + auto server = Server::ptr::Create(CreateWith{domain}, id, + std::vector{}, registry); + cloud->AddServer(server); + cloud->servers().at(id).priority = priority; + return server; + } + + TestContext context; + AeContext ae_context; + MapDomainStorage storage; + Domain domain; + Aether::ptr aether; + AdapterRegistry::ptr registry; + RegistrationCloud::ptr cloud; +}; + +void test_SingleRegistrationServerSelectsPriorityZero() { + Fixture f; + f.AddServer(ServerId{0}, 0); + + RootServerSelectStream stream{f.ae_context, f.cloud.Load()}; + int cloud_errors = 0; + stream.cloud_error_event().Subscribe([&]() { ++cloud_errors; }); + + TEST_ASSERT_EQUAL_UINT( + 1, RootServerSelectStreamTestAccess::ServerPriority(stream)); + RootServerSelectStreamTestAccess::ServerError(stream); + TEST_ASSERT_EQUAL_INT(1, cloud_errors); +} + +void test_RegistrationServersFailOverByNextPriority() { + Fixture f; + f.AddServer(ServerId{80}, 1); + f.AddServer(ServerId{2}, 0); + + RootServerSelectStream stream{f.ae_context, f.cloud.Load()}; + int cloud_errors = 0; + stream.cloud_error_event().Subscribe([&]() { ++cloud_errors; }); + + TEST_ASSERT_EQUAL_UINT( + 1, RootServerSelectStreamTestAccess::ServerPriority(stream)); + RootServerSelectStreamTestAccess::ServerError(stream); + TEST_ASSERT_EQUAL_UINT( + 2, RootServerSelectStreamTestAccess::ServerPriority(stream)); + + RootServerSelectStreamTestAccess::ServerError(stream); + TEST_ASSERT_EQUAL_INT(1, cloud_errors); +} + +} // namespace test_registration_root_server_select +} // namespace ae + +int run_test_registration_root_server_select() { + using namespace ae::test_registration_root_server_select; // NOLINT + + UNITY_BEGIN(); + RUN_TEST(test_SingleRegistrationServerSelectsPriorityZero); + RUN_TEST(test_RegistrationServersFailOverByNextPriority); + return UNITY_END(); +} + +#endif // AE_SUPPORT_REGISTRATION