Add PackagePtr::Start for FastTx package send

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")
 

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<AeMessage, AeMessage> {
         }},
         request_policy_);
   }
+
+  std::optional<prepared_packet::PreparedSendMessageBlock>
+  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<prepared_packet::PreparedSendMessageBlock>
+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});

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 <cstdint>
+#include <optional>
+
 #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<prepared_packet::PreparedSendMessageBlock>
+  ExportPreparedSendMessageBlock(std::uint32_t reserve_nonce_count);
+
  private:
   void ConnectReceive();
   void ConnectSend();

aether/prepared_packet/package.cpp

@@ -0,0 +1,93 @@
+#include "aether/prepared_packet/package.h"
+
+#include <utility>
+
+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<std::uint8_t const>{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

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 <cstddef>
+#include <cstdint>
+#include <optional>
+#include <utility>
+
+#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<std::optional<std::size_t>(
+      PreparedEndpoint const& endpoint, Span<std::uint8_t const> 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> package) noexcept
+      : package_{std::move(package)} {}
+
+  template <typename... TArgs>
+  explicit PackagePtr(AeContext const& ae_context, TArgs&&... args)
+      : package_{MakeRcPtr<Package>(ae_context, std::forward<TArgs>(args)...)} {
+  }
+
+  Package& operator*() const { return *package_; }
+  Package* operator->() const { return package_.get(); }
+  explicit operator bool() const noexcept {
+    return static_cast<bool>(package_);
+  }
+
+  void Reset() noexcept { package_.Reset(); }
+  RcPtr<Package> const& get() const noexcept { return package_; }
+
+ private:
+  RcPtr<Package> 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_

aether/prepared_packet/packet_encoder.cpp

@@ -0,0 +1,73 @@
+#include "aether/prepared_packet/packet_encoder.h"
+
+#include <memory>
+#include <utility>
+
+#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<PreparedSendMessageKeyProvider>(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<AuthorizedApi>{
+          [&block, &payload](ApiContext<AuthorizedApi>& 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

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_