From 7e2b6c981b67586af7b1348181296c63cb0ed70d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 20 Jul 2026 23:31:18 +0000 Subject: [PATCH 1/2] Add PackagePtr with Start() for FastTx send path. Bring prepared_packet encode/export foundation onto main and add Package / PackagePtr so EncodePacket output can be owned and started through an externally owned UDP send function, with would-block retry via the task scheduler. Co-authored-by: Nikolay Chirkov --- aether/CMakeLists.txt | 4 + aether/client_messages/p2p_message_stream.cpp | 35 ++++ aether/client_messages/p2p_message_stream.h | 7 + aether/prepared_packet/package.cpp | 93 +++++++++ aether/prepared_packet/package.h | 114 +++++++++++ aether/prepared_packet/packet_encoder.cpp | 73 +++++++ aether/prepared_packet/packet_encoder.h | 21 +++ .../prepared_packet/prepare_send_message.cpp | 28 +++ aether/prepared_packet/prepare_send_message.h | 37 ++++ .../prepared_packet/prepared_send_message.cpp | 41 ++++ .../prepared_packet/prepared_send_message.h | 94 +++++++++ .../client_server_connection.cpp | 55 ++++++ .../client_server_connection.h | 9 + tests/CMakeLists.txt | 1 + tests/test-prepared-packet/CMakeLists.txt | 33 ++++ tests/test-prepared-packet/main.cpp | 34 ++++ .../test-prepared-packet/test-package-ptr.cpp | 178 ++++++++++++++++++ 17 files changed, 857 insertions(+) create mode 100644 aether/prepared_packet/package.cpp create mode 100644 aether/prepared_packet/package.h create mode 100644 aether/prepared_packet/packet_encoder.cpp create mode 100644 aether/prepared_packet/packet_encoder.h create mode 100644 aether/prepared_packet/prepare_send_message.cpp create mode 100644 aether/prepared_packet/prepare_send_message.h create mode 100644 aether/prepared_packet/prepared_send_message.cpp create mode 100644 aether/prepared_packet/prepared_send_message.h create mode 100644 tests/test-prepared-packet/CMakeLists.txt create mode 100644 tests/test-prepared-packet/main.cpp create mode 100644 tests/test-prepared-packet/test-package-ptr.cpp diff --git a/aether/CMakeLists.txt b/aether/CMakeLists.txt index 13a303d1..0dd69727 100644 --- a/aether/CMakeLists.txt +++ b/aether/CMakeLists.txt @@ -198,6 +198,10 @@ list(APPEND aether_srcs list(APPEND aether_srcs "server_connections/client_server_connection.cpp" + "prepared_packet/prepared_send_message.cpp" + "prepared_packet/prepare_send_message.cpp" + "prepared_packet/packet_encoder.cpp" + "prepared_packet/package.cpp" "server_connections/channel_connection.cpp" "server_connections/server_connection.cpp") diff --git a/aether/client_messages/p2p_message_stream.cpp b/aether/client_messages/p2p_message_stream.cpp index c202645f..23ac7ecf 100644 --- a/aether/client_messages/p2p_message_stream.cpp +++ b/aether/client_messages/p2p_message_stream.cpp @@ -25,6 +25,7 @@ #include "aether/cloud_connections/cloud_request.h" #include "aether/cloud_connections/cloud_subscription.h" +#include "aether/cloud_connections/cloud_server_connection.h" #include "aether/client_messages/client_messages_tele.h" @@ -48,6 +49,30 @@ class MessageSendStream final : public IStream { }}, request_policy_); } + + std::optional + ExportPreparedSendMessageBlock(Uid target_uid, + std::uint32_t reserve_nonce_count) { + for (auto* sc : cloud_connection_->servers()) { + if (sc == nullptr) { + continue; + } + + auto* conn = sc->client_connection(); + if (conn == nullptr) { + continue; + } + + auto block = conn->ExportPreparedSendMessageBlock(target_uid, + reserve_nonce_count); + if (block) { + return block; + } + } + + return std::nullopt; + } + StreamInfo stream_info() const override { return stream_info_; } OutDataEvent::Subscriber out_data_event() override { return out_data_event_; } StreamUpdateEvent::Subscriber stream_update_event() override { @@ -203,6 +228,16 @@ void P2pStream::WriteOut(DataBuffer const& data) { Uid const& P2pStream::destination() const { return destination_; } +std::optional +P2pStream::ExportPreparedSendMessageBlock(std::uint32_t reserve_nonce_count) { + if (message_send_stream_ == nullptr) { + return std::nullopt; + } + + return message_send_stream_->ExportPreparedSendMessageBlock( + destination_, reserve_nonce_count); +} + void P2pStream::ConnectReceive() { out_data_sub_ = handle_.out_data_event().Subscribe(MethodPtr<&P2pStream::WriteOut>{this}); diff --git a/aether/client_messages/p2p_message_stream.h b/aether/client_messages/p2p_message_stream.h index 64da048b..e243174f 100644 --- a/aether/client_messages/p2p_message_stream.h +++ b/aether/client_messages/p2p_message_stream.h @@ -17,6 +17,9 @@ #ifndef AETHER_CLIENT_MESSAGES_P2P_MESSAGE_STREAM_H_ #define AETHER_CLIENT_MESSAGES_P2P_MESSAGE_STREAM_H_ +#include +#include + #include "aether/common.h" #include "aether/ae_context.h" @@ -28,6 +31,7 @@ #include "aether/client_messages/p2p_port_handle.h" #include "aether/cloud_connections/cloud_server_connections.h" #include "aether/connection_manager/client_cloud_manager.h" +#include "aether/prepared_packet/prepared_send_message.h" namespace ae { class Client; @@ -54,6 +58,9 @@ class P2pStream final : public ByteIStream { void WriteOut(DataBuffer const& data); Uid const& destination() const; + std::optional + ExportPreparedSendMessageBlock(std::uint32_t reserve_nonce_count); + private: void ConnectReceive(); void ConnectSend(); diff --git a/aether/prepared_packet/package.cpp b/aether/prepared_packet/package.cpp new file mode 100644 index 00000000..98874fe0 --- /dev/null +++ b/aether/prepared_packet/package.cpp @@ -0,0 +1,93 @@ +#include "aether/prepared_packet/package.h" + +#include + +namespace ae::prepared_packet { + +Package::Package(AeContext const& ae_context, PreparedEndpoint endpoint, + DataBuffer&& data) + : ae_context_{ae_context}, + endpoint_{endpoint}, + data_{std::move(data)} {} + +void Package::Start(SendFunc&& send) { + if (started_ || done_ || is_finished()) { + return; + } + + started_ = true; + send_ = std::move(send); + TrySend(); +} + +void Package::Stop() noexcept { + if (done_ || is_finished()) { + return; + } + + done_ = true; + retry_sub_.Reset(); + send_ = {}; + WriteAction::SetStatus(Status::kStop); +} + +void Package::TrySend() { + if (done_ || is_finished()) { + return; + } + + if (!send_) { + done_ = true; + WriteAction::SetStatus(Status::kFail); + return; + } + + auto const bytes = + Span{data_.data(), data_.size()}; + auto const sent = send_(endpoint_, bytes); + if (!sent) { + done_ = true; + retry_sub_.Reset(); + send_ = {}; + WriteAction::SetStatus(Status::kFail); + return; + } + + if (*sent == 0) { + // Would-block: retry on the task scheduler without busy looping. + if (!retry_sub_) { + retry_sub_ = ae_context_.scheduler().Task([this]() { + retry_sub_.Reset(); + TrySend(); + }); + } + return; + } + + if (*sent != data_.size()) { + done_ = true; + retry_sub_.Reset(); + send_ = {}; + WriteAction::SetStatus(Status::kFail); + return; + } + + done_ = true; + retry_sub_.Reset(); + send_ = {}; + WriteAction::SetStatus(Status::kSuccess); +} + +PackagePtr MakePackage(AeContext const& ae_context, + PreparedSendMessageBlock& block, + DataBuffer const& payload) { + DataBuffer encoded; + auto const result = EncodePacket(block, payload, encoded); + if (!result) { + return PackagePtr{}; + } + + return PackagePtr{ae_context, block.endpoint, std::move(encoded)}; +} + +} // namespace ae::prepared_packet diff --git a/aether/prepared_packet/package.h b/aether/prepared_packet/package.h new file mode 100644 index 00000000..bb2166fe --- /dev/null +++ b/aether/prepared_packet/package.h @@ -0,0 +1,114 @@ +/* + * FastTx encoded package ownership and send start. + * + * Package holds encoded packet bytes and the destination endpoint selected + * during PrepareSendMessage. It does not own sockets, DNS, or connections — + * Start() sends through a caller-provided raw send function. + */ +#ifndef AETHER_PREPARED_PACKET_PACKAGE_H_ +#define AETHER_PREPARED_PACKET_PACKAGE_H_ + +#include +#include +#include +#include + +#include "aether-miscpp/types/small_function.h" + +#include "aether/ae_context.h" +#include "aether/common.h" +#include "aether/ptr/rc_ptr.h" +#include "aether/types/data_buffer.h" +#include "aether/types/span.h" +#include "aether/write_action/write_action.h" + +#include "aether/prepared_packet/packet_encoder.h" +#include "aether/prepared_packet/prepared_send_message.h" + +namespace ae::prepared_packet { + +/** + * \brief Encoded send_message package ready for external UDP send. + */ +class Package final : public WriteAction { + public: + /** + * \brief Raw send attempt. + * + * - nullopt: hard failure + * - 0: not sent yet (would-block); Start may retry + * - >0: bytes accepted by transport (must match package size) + */ + using SendFunc = SmallFunction( + PreparedEndpoint const& endpoint, Span data)>; + + Package(AeContext const& ae_context, PreparedEndpoint endpoint, + DataBuffer&& data); + + AE_CLASS_MOVE_ONLY(Package) + + PreparedEndpoint const& endpoint() const noexcept { return endpoint_; } + DataBuffer const& data() const noexcept { return data_; } + bool is_started() const noexcept { return started_; } + + /** + * \brief Begin sending through the externally owned transport. + * + * Safe to call once. After completion, further Start calls are ignored. + */ + void Start(SendFunc&& send); + + void Stop() noexcept override; + + private: + void TrySend(); + + AeContext ae_context_; + PreparedEndpoint endpoint_; + DataBuffer data_; + SendFunc send_; + bool started_ = false; + bool done_ = false; + TaskSubscription retry_sub_; +}; + +/** + * \brief Shared ownership of a FastTx Package (ActionPtr-shaped for encode→send). + */ +class PackagePtr { + public: + PackagePtr() noexcept = default; + explicit PackagePtr(std::nullptr_t) noexcept : PackagePtr() {} + explicit PackagePtr(RcPtr package) noexcept + : package_{std::move(package)} {} + + template + explicit PackagePtr(AeContext const& ae_context, TArgs&&... args) + : package_{MakeRcPtr(ae_context, std::forward(args)...)} { + } + + Package& operator*() const { return *package_; } + Package* operator->() const { return package_.get(); } + explicit operator bool() const noexcept { + return static_cast(package_); + } + + void Reset() noexcept { package_.Reset(); } + RcPtr const& get() const noexcept { return package_; } + + private: + RcPtr package_; +}; + +/** + * \brief Encode payload with prepared block and wrap the result in PackagePtr. + * + * Returns empty PackagePtr when encoding fails (e.g. nonce exhausted). + */ +PackagePtr MakePackage(AeContext const& ae_context, + PreparedSendMessageBlock& block, + DataBuffer const& payload); + +} // namespace ae::prepared_packet + +#endif // AETHER_PREPARED_PACKET_PACKAGE_H_ diff --git a/aether/prepared_packet/packet_encoder.cpp b/aether/prepared_packet/packet_encoder.cpp new file mode 100644 index 00000000..0c87e7bb --- /dev/null +++ b/aether/prepared_packet/packet_encoder.cpp @@ -0,0 +1,73 @@ +#include "aether/prepared_packet/packet_encoder.h" + +#include +#include + +#include "aether/crypto/ikey_provider.h" +#include "aether/crypto/sync_crypto_provider.h" + +#include "aether/api_protocol/api_context.h" +#include "aether/api_protocol/sub_api.h" + +#include "aether/work_cloud_api/ae_message.h" +#include "aether/work_cloud_api/work_server_api/authorized_api.h" +#include "aether/work_cloud_api/work_server_api/login_api.h" + +namespace ae::prepared_packet { +namespace { + +class PreparedSendMessageKeyProvider final : public ISyncKeyProvider { + public: + explicit PreparedSendMessageKeyProvider(PreparedSendMessageBlock& block) + : block_{&block} {} + + Key GetKey() const override { + return block_->client_to_server_key; + } + + CryptoNonce const& Nonce() const override { + return block_->next_nonce; + } + + private: + PreparedSendMessageBlock* block_; +}; + +} // namespace + +EncodePacketResult EncodePacket(PreparedSendMessageBlock& block, + DataBuffer const& payload, + DataBuffer& out) { + if (block.nonce_left == 0) { + out.clear(); + return EncodePacketResult{EncodePacketError::kNonceExhausted, 0}; + } + + // Match the existing ClientKeyProvider semantics: + // consume next nonce before encryption. + block.next_nonce.Next(); + --block.nonce_left; + + auto key_provider = + std::make_unique(block); + SyncEncryptProvider encrypt_provider{std::move(key_provider)}; + + ProtocolContext protocol_context; + LoginApi login_api{protocol_context, encrypt_provider}; + + auto api_context = ApiContext{login_api}; + + api_context->login_by_alias( + block.sender_ephemeral_uid, + SubApi{ + [&block, &payload](ApiContext& auth_api) { + auth_api->send_message( + AeMessage{block.target_uid, DataBuffer{payload}}); + }}); + + out = std::move(api_context).Pack(); + + return EncodePacketResult{EncodePacketError::kNone, out.size()}; +} + +} // namespace ae::prepared_packet diff --git a/aether/prepared_packet/packet_encoder.h b/aether/prepared_packet/packet_encoder.h new file mode 100644 index 00000000..6f9d7371 --- /dev/null +++ b/aether/prepared_packet/packet_encoder.h @@ -0,0 +1,21 @@ +/* + * Prepared packet encoder. + * + * EncodePacket only builds Aether packet bytes and advances the reserved nonce + * range. It does not send, open sockets, resolve DNS, or know platform + * transport. + */ +#ifndef AETHER_PREPARED_PACKET_PACKET_ENCODER_H_ +#define AETHER_PREPARED_PACKET_PACKET_ENCODER_H_ + +#include "aether/prepared_packet/prepared_send_message.h" + +namespace ae::prepared_packet { + +EncodePacketResult EncodePacket(PreparedSendMessageBlock& block, + DataBuffer const& payload, + DataBuffer& out); + +} // namespace ae::prepared_packet + +#endif // AETHER_PREPARED_PACKET_PACKET_ENCODER_H_ diff --git a/aether/prepared_packet/prepare_send_message.cpp b/aether/prepared_packet/prepare_send_message.cpp new file mode 100644 index 00000000..2a56800f --- /dev/null +++ b/aether/prepared_packet/prepare_send_message.cpp @@ -0,0 +1,28 @@ +/* + * 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/prepared_packet/prepare_send_message.h" + +#include "aether/client_messages/p2p_message_stream.h" + +namespace ae::prepared_packet { + +std::optional PrepareSendMessage( + P2pStream& stream, std::uint32_t reserve_nonce_count) { + return stream.ExportPreparedSendMessageBlock(reserve_nonce_count); +} + +} // namespace ae::prepared_packet diff --git a/aether/prepared_packet/prepare_send_message.h b/aether/prepared_packet/prepare_send_message.h new file mode 100644 index 00000000..08d0e710 --- /dev/null +++ b/aether/prepared_packet/prepare_send_message.h @@ -0,0 +1,37 @@ +/* + * 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_PREPARED_PACKET_PREPARE_SEND_MESSAGE_H_ +#define AETHER_PREPARED_PACKET_PREPARE_SEND_MESSAGE_H_ + +#include +#include + +#include "aether/prepared_packet/prepared_send_message.h" + +namespace ae { + +class P2pStream; + +namespace prepared_packet { + +std::optional PrepareSendMessage( + P2pStream& stream, std::uint32_t reserve_nonce_count); + +} // namespace prepared_packet +} // namespace ae + +#endif // AETHER_PREPARED_PACKET_PREPARE_SEND_MESSAGE_H_ diff --git a/aether/prepared_packet/prepared_send_message.cpp b/aether/prepared_packet/prepared_send_message.cpp new file mode 100644 index 00000000..db965ee6 --- /dev/null +++ b/aether/prepared_packet/prepared_send_message.cpp @@ -0,0 +1,41 @@ +#include "aether/prepared_packet/prepared_send_message.h" + +#include + +namespace ae::prepared_packet { + +std::optional MakePreparedEndpoint(Endpoint const& endpoint) { + PreparedEndpoint candidate; + candidate.protocol = endpoint.protocol; + candidate.port = endpoint.port; + + bool is_ip = false; + + std::visit( + [&](auto const& addr) { + using T = std::decay_t; + + if constexpr (std::is_same_v) { + candidate.version = PreparedIpVersion::kIpV4; + for (std::size_t i = 0; i < 4; ++i) { + candidate.ip[i] = addr.ipv4_value[i]; + } + is_ip = true; + } else if constexpr (std::is_same_v) { + candidate.version = PreparedIpVersion::kIpV6; + for (std::size_t i = 0; i < 16; ++i) { + candidate.ip[i] = addr.ipv6_value[i]; + } + is_ip = true; + } + }, + endpoint.address); + + if (!is_ip) { + return std::nullopt; + } + + return candidate; +} + +} // namespace ae::prepared_packet diff --git a/aether/prepared_packet/prepared_send_message.h b/aether/prepared_packet/prepared_send_message.h new file mode 100644 index 00000000..fff70874 --- /dev/null +++ b/aether/prepared_packet/prepared_send_message.h @@ -0,0 +1,94 @@ +/* + * Prepared send_message block. + * + * This is transport-neutral state for encoding a send_message packet. + * It may contain an endpoint selected by the full Aether client, but it does + * not own sockets, DNS, connections, channels, or timers. + */ +#ifndef AETHER_PREPARED_PACKET_PREPARED_SEND_MESSAGE_H_ +#define AETHER_PREPARED_PACKET_PREPARED_SEND_MESSAGE_H_ + +#include +#include +#include +#include + +#include "aether/types/address.h" +#include "aether/types/data_buffer.h" +#include "aether/types/uid.h" + +#include "aether/crypto/key.h" +#include "aether/crypto/crypto_nonce.h" + +namespace ae::prepared_packet { +static constexpr std::uint32_t kMagic = 0x50534456; // "PSDV" +static constexpr std::uint32_t kVersion = 1; +// Serialized PreparedSendMessageBlock is a few hundred bytes. Keep the RTC +// footprint small enough for ESP32 RTC slow memory (8 KiB on ESP32-C6). +static constexpr std::size_t kMaxPreparedBlockBytes = 512; + +struct RetainedPreparedBlock { + std::uint32_t magic; + std::uint32_t version; + std::uint32_t size; + std::uint32_t checksum; + std::array bytes; +}; + +enum class PreparedIpVersion : std::uint8_t { + kIpV4 = 4, + kIpV6 = 6, +}; + +struct PreparedEndpoint { + AE_REFLECT_MEMBERS(version, protocol, port, ip) + PreparedIpVersion version = PreparedIpVersion::kIpV4; + Protocol protocol = Protocol::kUdp; + std::uint16_t port = 0; + + // IPv4 uses first 4 bytes. + // IPv6 uses all 16 bytes. + std::array ip{}; +}; + +struct PreparedSendMessageBlock { + AE_REFLECT_MEMBERS(endpoint, sender_ephemeral_uid, target_uid, + client_to_server_key, next_nonce, nonce_left) + PreparedEndpoint endpoint; + + Uid sender_ephemeral_uid; + Uid target_uid; + + Key client_to_server_key; + + CryptoNonce next_nonce; + std::uint32_t nonce_left = 0; +}; + +enum class EncodePacketError { + kNone = 0, + kNonceExhausted, +}; + +inline char const* ToString(EncodePacketError error) { + switch (error) { + case EncodePacketError::kNone: + return "none"; + case EncodePacketError::kNonceExhausted: + return "nonce_exhausted"; + } + return "unknown"; +} + +struct EncodePacketResult { + EncodePacketError error = EncodePacketError::kNone; + std::size_t bytes_written = 0; + + explicit operator bool() const { return error == EncodePacketError::kNone; } +}; + +std::optional MakePreparedEndpoint(Endpoint const& endpoint); + +} // namespace ae::prepared_packet + +#endif // AETHER_PREPARED_PACKET_PREPARED_SEND_MESSAGE_H_ diff --git a/aether/server_connections/client_server_connection.cpp b/aether/server_connections/client_server_connection.cpp index 66d29b46..bf173e8e 100644 --- a/aether/server_connections/client_server_connection.cpp +++ b/aether/server_connections/client_server_connection.cpp @@ -142,6 +142,7 @@ ClientServerConnection::ClientServerConnection(AeContext const& ae_context, Ptr const& client, Ptr const& server) : ae_context_{ae_context}, + client_{client}, server_{server}, ephemeral_uid_{client->ephemeral_uid()}, crypto_provider_{std::make_unique< @@ -197,6 +198,60 @@ ServerConnection& ClientServerConnection::server_connection() { return server_connection_.server_connection; } +std::optional +ClientServerConnection::ExportPreparedSendMessageBlock( + Uid target_uid, std::uint32_t reserve_nonce_count) { + auto client = client_.Lock(); + auto server = server_.Lock(); + + if ((client == nullptr) || (server == nullptr) || + (reserve_nonce_count == 0)) { + return std::nullopt; + } + + std::optional prepared_endpoint; + + for (auto const& e : server->endpoints) { + if (e.protocol != Protocol::kUdp) { + continue; + } + + auto endpoint = prepared_packet::MakePreparedEndpoint(e); + if (endpoint) { + prepared_endpoint = *endpoint; + break; + } + } + + if (!prepared_endpoint) { + return std::nullopt; + } + + auto* server_key = client->server_state(server->server_id); + if (server_key == nullptr) { + return std::nullopt; + } + + prepared_packet::PreparedSendMessageBlock block; + block.endpoint = *prepared_endpoint; + block.sender_ephemeral_uid = ephemeral_uid_; + block.target_uid = target_uid; + block.client_to_server_key = server_key->client_to_server(); + + // Store current nonce. EncodePacket() will call Next() before encryption, + // matching existing ClientKeyProvider behavior. + block.next_nonce = server_key->nonce(); + block.nonce_left = reserve_nonce_count; + + // Burn/reserve the same nonce range in the full client, + // so the normal Aether path cannot reuse it later. + for (std::uint32_t i = 0; i < reserve_nonce_count; ++i) { + server_key->Next(); + } + + return block; +} + void ClientServerConnection::OutData(DataBuffer const& data) { auto parser = ApiParser{protocol_context_, data}; parser.Parse(client_api_unsafe_); diff --git a/aether/server_connections/client_server_connection.h b/aether/server_connections/client_server_connection.h index 83a0aac2..d1c54796 100644 --- a/aether/server_connections/client_server_connection.h +++ b/aether/server_connections/client_server_connection.h @@ -17,6 +17,9 @@ #ifndef AETHER_SERVER_CONNECTIONS_CLIENT_SERVER_CONNECTION_H_ #define AETHER_SERVER_CONNECTIONS_CLIENT_SERVER_CONNECTION_H_ +#include +#include + #include "aether/common.h" #include "aether/ae_context.h" #include "aether/crypto/icrypto_provider.h" @@ -27,6 +30,7 @@ #include "aether/work_cloud_api/client_api/client_api_unsafe.h" #include "aether/work_cloud_api/work_server_api/authorized_api.h" +#include "aether/prepared_packet/prepared_send_message.h" #include "aether/server_connections/server_connection.h" namespace ae { @@ -74,10 +78,15 @@ class ClientServerConnection { ServerConnection& server_connection(); + std::optional + ExportPreparedSendMessageBlock(Uid target_uid, + std::uint32_t reserve_nonce_count); + private: void OutData(DataBuffer const& data); AeContext ae_context_; + PtrView client_; PtrView server_; Uid ephemeral_uid_; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 21fae1e6..3c62bfa2 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -45,5 +45,6 @@ add_subdirectory(test-ptr) add_subdirectory(test-domain-storage) add_subdirectory(test-serial-port) add_subdirectory(test-tasks) +add_subdirectory(test-prepared-packet) add_subdirectory(third_party_tests) diff --git a/tests/test-prepared-packet/CMakeLists.txt b/tests/test-prepared-packet/CMakeLists.txt new file mode 100644 index 00000000..63822cbd --- /dev/null +++ b/tests/test-prepared-packet/CMakeLists.txt @@ -0,0 +1,33 @@ +# 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. + +cmake_minimum_required(VERSION 3.16) + +list(APPEND test_srcs + main.cpp + test-package-ptr.cpp +) + +if(NOT CM_PLATFORM) + project(test-prepared-packet LANGUAGES CXX) + + add_executable(${PROJECT_NAME}) + target_sources(${PROJECT_NAME} PRIVATE ${test_srcs}) + target_include_directories(${PROJECT_NAME} PRIVATE ${ROOT_DIR}) + target_link_libraries(${PROJECT_NAME} PRIVATE aether unity gcem) + + add_test(NAME ${PROJECT_NAME} COMMAND $) +else() + message(WARNING "Not implemented for ${CM_PLATFORM}") +endif() diff --git a/tests/test-prepared-packet/main.cpp b/tests/test-prepared-packet/main.cpp new file mode 100644 index 00000000..32d33f6d --- /dev/null +++ b/tests/test-prepared-packet/main.cpp @@ -0,0 +1,34 @@ +/* + * 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 + +void setUp() {} +void tearDown() {} + +extern void test_package_ptr_start_success(); +extern void test_package_ptr_start_retry_then_success(); +extern void test_package_ptr_start_fail(); +extern void test_package_ptr_stop(); + +int main() { + UNITY_BEGIN(); + RUN_TEST(test_package_ptr_start_success); + RUN_TEST(test_package_ptr_start_retry_then_success); + RUN_TEST(test_package_ptr_start_fail); + RUN_TEST(test_package_ptr_stop); + return UNITY_END(); +} diff --git a/tests/test-prepared-packet/test-package-ptr.cpp b/tests/test-prepared-packet/test-package-ptr.cpp new file mode 100644 index 00000000..dcc44c9d --- /dev/null +++ b/tests/test-prepared-packet/test-package-ptr.cpp @@ -0,0 +1,178 @@ +/* + * Copyright 2026 Aethernet Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include + +#include +#include +#include +#include + +#include "aether/ae_context.h" +#include "aether/prepared_packet/package.h" +#include "aether/types/data_buffer.h" + +namespace ae::test_package_ptr { +namespace { + +struct TestContext { + AeCtx ToAeContext() const { + static constexpr auto table = + AeCtxTable{nullptr, [](void* obj) -> TaskScheduler& { + return static_cast(obj)->sched; + }}; + return AeCtx{ + const_cast(this), // NOLINT + &table, + }; + } + + TaskScheduler sched; +}; + +prepared_packet::PreparedEndpoint MakeEndpoint() { + prepared_packet::PreparedEndpoint endpoint; + endpoint.version = prepared_packet::PreparedIpVersion::kIpV4; + endpoint.protocol = Protocol::kUdp; + endpoint.port = 9000; + endpoint.ip = {127, 0, 0, 1}; + return endpoint; +} + +DataBuffer MakePayload() { return DataBuffer{1, 2, 3, 4, 5}; } + +} // namespace + +void test_package_ptr_start_success() { + auto ctx = TestContext{}; + auto payload = MakePayload(); + auto package = prepared_packet::PackagePtr{AeContext{ctx}, MakeEndpoint(), + DataBuffer{payload}}; + + TEST_ASSERT_TRUE(static_cast(package)); + TEST_ASSERT_FALSE(package->is_started()); + + auto status = WriteAction::Status::kStop; + auto status_sub = package->status_event().Subscribe( + [&](WriteAction::Status s) { status = s; }); + + std::size_t send_calls = 0; + package->Start([&](prepared_packet::PreparedEndpoint const& endpoint, + Span data) + -> std::optional { + ++send_calls; + TEST_ASSERT_EQUAL_UINT16(9000, endpoint.port); + TEST_ASSERT_EQUAL_UINT32(payload.size(), data.size()); + return data.size(); + }); + + TEST_ASSERT_TRUE(package->is_started()); + TEST_ASSERT_EQUAL_UINT32(1, send_calls); + TEST_ASSERT_EQUAL(static_cast(WriteAction::Status::kSuccess), + static_cast(status)); + TEST_ASSERT_TRUE(package->is_finished()); +} + +void test_package_ptr_start_retry_then_success() { + auto ctx = TestContext{}; + auto payload = MakePayload(); + auto package = prepared_packet::PackagePtr{AeContext{ctx}, MakeEndpoint(), + DataBuffer{payload}}; + + auto status = WriteAction::Status::kStop; + auto status_sub = package->status_event().Subscribe( + [&](WriteAction::Status s) { status = s; }); + + std::size_t send_calls = 0; + package->Start([&](prepared_packet::PreparedEndpoint const&, + Span data) + -> std::optional { + ++send_calls; + if (send_calls == 1) { + return 0; + } + return data.size(); + }); + + TEST_ASSERT_EQUAL_UINT32(1, send_calls); + TEST_ASSERT_FALSE(package->is_finished()); + + ctx.sched.Update(); + + TEST_ASSERT_EQUAL_UINT32(2, send_calls); + TEST_ASSERT_EQUAL(static_cast(WriteAction::Status::kSuccess), + static_cast(status)); + TEST_ASSERT_TRUE(package->is_finished()); +} + +void test_package_ptr_start_fail() { + auto ctx = TestContext{}; + auto package = prepared_packet::PackagePtr{AeContext{ctx}, MakeEndpoint(), + MakePayload()}; + + auto status = WriteAction::Status::kStop; + auto status_sub = package->status_event().Subscribe( + [&](WriteAction::Status s) { status = s; }); + + package->Start([](prepared_packet::PreparedEndpoint const&, + Span) -> std::optional { + return std::nullopt; + }); + + TEST_ASSERT_EQUAL(static_cast(WriteAction::Status::kFail), + static_cast(status)); + TEST_ASSERT_TRUE(package->is_finished()); +} + +void test_package_ptr_stop() { + auto ctx = TestContext{}; + auto package = prepared_packet::PackagePtr{AeContext{ctx}, MakeEndpoint(), + MakePayload()}; + + auto status = WriteAction::Status::kSuccess; + auto status_sub = package->status_event().Subscribe( + [&](WriteAction::Status s) { status = s; }); + + std::size_t send_calls = 0; + package->Start([&](prepared_packet::PreparedEndpoint const&, + Span) -> std::optional { + ++send_calls; + return 0; + }); + + TEST_ASSERT_EQUAL_UINT32(1, send_calls); + package->Stop(); + + TEST_ASSERT_EQUAL(static_cast(WriteAction::Status::kStop), + static_cast(status)); + TEST_ASSERT_TRUE(package->is_finished()); + + ctx.sched.Update(); + TEST_ASSERT_EQUAL_UINT32(1, send_calls); +} + +} // namespace ae::test_package_ptr + +void test_package_ptr_start_success() { + ae::test_package_ptr::test_package_ptr_start_success(); +} +void test_package_ptr_start_retry_then_success() { + ae::test_package_ptr::test_package_ptr_start_retry_then_success(); +} +void test_package_ptr_start_fail() { + ae::test_package_ptr::test_package_ptr_start_fail(); +} +void test_package_ptr_stop() { ae::test_package_ptr::test_package_ptr_stop(); } From 63d678cc2ecb4b1a5c5b0a3f48996e7f5e1e2d13 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 20 Jul 2026 23:34:20 +0000 Subject: [PATCH 2/2] Prefer brace initialization for Package flags. Co-authored-by: Nikolay Chirkov --- aether/prepared_packet/package.h | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/aether/prepared_packet/package.h b/aether/prepared_packet/package.h index bb2166fe..f6f037fb 100644 --- a/aether/prepared_packet/package.h +++ b/aether/prepared_packet/package.h @@ -67,8 +67,8 @@ class Package final : public WriteAction { PreparedEndpoint endpoint_; DataBuffer data_; SendFunc send_; - bool started_ = false; - bool done_ = false; + bool started_{false}; + bool done_{false}; TaskSubscription retry_sub_; };