522 fix cloud connection freezing

.opencode/agents/tester.md

@@ -10,6 +10,7 @@ permission:
   glob: allow
   bash:
     "*": deny
+    "timeout *aether-client-cpp-cloud*": allow
     "rm -rf *state": allow
     "*aether-client-cpp-cloud*": allow
     "ninja test": allow

aether/ae_actions/ae_actions_tele.h

@@ -17,7 +17,7 @@
 #ifndef AETHER_AE_ACTIONS_AE_ACTIONS_TELE_H_
 #define AETHER_AE_ACTIONS_AE_ACTIONS_TELE_H_
 
-#include "aether/tele.h"
+#include "aether/tele.h"  // IWYU pragma: exports
 
 #if AE_SUPPORT_REGISTRATION
 AE_TELE_MODULE(kRegister, 1000, 1000, 1050);

aether/ae_actions/ping.cpp

@@ -82,11 +82,12 @@ void Ping::Start(TimePoint current_time) {
   assert(cc != nullptr && "Ping::Start requires a client connection");
   assert(cc->stream_info().link_state == LinkState::kLinked &&
          "Ping::Start requires linked connection");
-  assert(!started_ && "Ping::Start must be called only once");
-  if (started_) {
+  assert(state_ == RequestState::kCreated &&
+         "Ping::Start must be called only once");
+  if (state_ != RequestState::kCreated) {
     return;
   }
-  started_ = true;
+  state_ = RequestState::kPending;
 
   auto& write_action = cc->AuthorizedApiCall(
       SubApi{[this, current_time](ApiContext<AuthorizedApi>& auth_api) {
@@ -120,15 +121,30 @@ void Ping::Start(TimePoint current_time) {
         timeout_sub_ = ae_context_.scheduler().DelayedTask(
             [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);
+          state_ = RequestState::kFinished;
+          ResetRequestSubscriptions();
+          result_event_.Emit(PingResult{Error{5}});
+        }
       }});
 
   write_sub_ = write_action.status_event().Subscribe([this](auto status) {
     if (status == WriteAction::Status::kFail) {
+      if (state_ != RequestState::kPending) {
+        return;
+      }
       AE_TELE_ERROR(kPingWriteError, "Ping write error");
+      state_ = RequestState::kFinished;
       ResetRequestSubscriptions();
-      result_event_.Emit(Error{1});
+      result_event_.Emit(PingResult{Error{1}});
     }
   });
+  if (state_ != RequestState::kPending) {
+    write_sub_.Reset();
+  }
 
 #  if DEBUG
   cc->LoginApiCall(SubApi{[&](ApiContext<LoginApi>& api_call) {
@@ -151,28 +167,47 @@ void Ping::PingResponse(RequestId request_id) {
   auto ping_duration =
       std::chrono::duration_cast<Duration>(current_time - request_start_);
 
-  AE_TELED_DEBUG("Ping server id {} request {} received by {}", server_id_,
-                 request_id, ping_duration);
-  ResetRequestSubscriptions();
-  result_event_.Emit(Ok{ping_duration});
+  AE_TELED_DEBUG("Ping received server id {} request {} duration {}",
+                 server_id_, request_id, ping_duration);
+  if (state_ == RequestState::kPending) {
+    state_ = RequestState::kFinished;
+    ResetRequestSubscriptions();
+    result_event_.Emit(PingResult{Ok{ping_duration}});
+  } else if (state_ == RequestState::kTimedOut) {
+    wait_result_sub_.Reset();
+    state_ = RequestState::kFinished;
+    result_event_.Emit(PingResult{LateDuration{ping_duration}});
+  }
 }
 
 void Ping::PingResponseError(RequestId request_id, std::int32_t error_code) {
-  AE_TELE_ERROR(kPingResponseError, "Ping server id {} request {} error {}",
-                server_id_, request_id, error_code);
-  ResetRequestSubscriptions();
-  if (error_code == -1) {
-    result_event_.Emit(Error{4});
-  } else {
-    result_event_.Emit(Error{3});
+  AE_TELE_ERROR(kPingResponseError,
+                "Ping error server id {} request {} code {}", server_id_,
+                request_id, error_code);
+  if (state_ == RequestState::kPending) {
+    state_ = RequestState::kFinished;
+    ResetRequestSubscriptions();
+    if (error_code == -1) {
+      result_event_.Emit(PingResult{Error{4}});
+    } else {
+      result_event_.Emit(PingResult{Error{3}});
+    }
+  } else if (state_ == RequestState::kTimedOut) {
+    wait_result_sub_.Reset();
+    state_ = RequestState::kFinished;
   }
 }
 
 void Ping::PingResponseTimeout(RequestId request_id) {
-  AE_TELE_ERROR(kPingTimeoutError, "Ping server id {} request {} timeout",
+  if (state_ != RequestState::kPending) {
+    return;
+  }
+  AE_TELE_ERROR(kPingTimeoutError, "Ping timeout server id {} request {}",
                 server_id_, request_id);
-  ResetRequestSubscriptions();
-  result_event_.Emit(Error{2});
+  state_ = RequestState::kTimedOut;
+  timeout_sub_.Reset();
+  write_sub_.Reset();
+  result_event_.Emit(PingResult{Error{2}});
 }
 
 void Ping::ResetRequestSubscriptions() {

aether/ae_actions/ping.h

@@ -21,6 +21,8 @@
 
 #if AE_ENABLE_PING
 
+#  include <variant>
+
 #  include "aether-miscpp/types/result.h"
 
 #  include "aether/ae_context.h"
@@ -36,7 +38,11 @@ class CloudServerConnection;
 
 class Ping {
  public:
-  using ResultEvent = Event<void(Result<Duration, int>)>;
+  struct LateDuration {
+    Duration duration;
+  };
+  using PingResult = std::variant<Ok<Duration>, LateDuration, Error<int>>;
+  using ResultEvent = Event<void(PingResult const&)>;
 
   Ping(AeContext const& ae_context,
        CloudServerConnection& cloud_server_connection, Duration next_ping_hint,
@@ -54,6 +60,13 @@ class Ping {
   void PingResponseTimeout(RequestId request_id);
   void ResetRequestSubscriptions();
 
+  enum class RequestState : char {
+    kCreated,
+    kPending,
+    kTimedOut,
+    kFinished,
+  };
+
   AeContext ae_context_;
   CloudServerConnection* cloud_server_connection_;
   Duration next_ping_hint_;
@@ -67,7 +80,7 @@ class Ping {
   Subscription write_sub_;
 
   ResultEvent result_event_;
-  bool started_{};
+  RequestState state_{RequestState::kCreated};
 };
 }  // namespace ae
 #endif  // AE_ENABLE_PING

aether/client_messages/client_messages_tele.h

@@ -17,7 +17,7 @@
 #ifndef AETHER_CLIENT_MESSAGES_CLIENT_MESSAGES_TELE_H_
 #define AETHER_CLIENT_MESSAGES_CLIENT_MESSAGES_TELE_H_
 
-#include "aether/tele.h"
+#include "aether/tele.h"  // IWYU pragma: exports
 
 AE_TELE_MODULE(kClientMessages, 8, 166, 170);
 

aether/client_messages/p2p_message_stream_manager.cpp

@@ -17,10 +17,12 @@
 #include "aether/client_messages/p2p_message_stream_manager.h"
 
 #include <cassert>
+#include <cstdint>
 #include <utility>
 
 #include "aether/client.h"
 #include "aether/client_messages/client_messages_tele.h"
+#include "aether/cloud_connections/cloud_server_connection.h"
 #include "aether/work_cloud_api/client_api/client_api_safe.h"
 
 namespace ae {
@@ -31,12 +33,13 @@ P2pMessageStreamManager::P2pMessageStreamManager(AeContext const& ae_context,
       client_{client},
       cloud_connection_{&client->cloud_connection()},
       on_message_received_sub_{CloudEventListener{
-          ApiEventSubscriber{[this](ClientApiSafe& client_api, auto*) {
+          ApiEventSubscriber{[this](ClientApiSafe& client_api,
+                                    [[maybe_unused]] CloudServerConnection* server_connection)
+                                 -> EventHandlerDeleter {
             return client_api.send_message_event().Subscribe(
                 MethodPtr<&P2pMessageStreamManager::NewMessageReceived>{this});
           }},
-          *cloud_connection_,
-          RequestPolicy::Replica{cloud_connection_->count_connections()}}} {}
+          *cloud_connection_, RequestPolicy::All{}}} {}
 
 P2pPortHandle P2pMessageStreamManager::CreatePort(Uid const& destination) {
   auto [port, is_new] = GetOrCreatePort(destination);
@@ -68,7 +71,6 @@ P2pMessageStreamManager::GetOrCreatePort(Uid const& destination) {
 
 void P2pMessageStreamManager::NewMessageReceived(AeMessage const& message) {
   AE_TELED_DEBUG("New message received {}", message.uid);
-
   auto [port, is_new] = GetOrCreatePort(message.uid);
   assert(port != nullptr);
 

aether/cloud_connections/cloud_connections_tele.h

@@ -17,7 +17,7 @@
 #ifndef AETHER_CLOUD_CONNECTIONS_CLOUD_CONNECTIONS_TELE_H_
 #define AETHER_CLOUD_CONNECTIONS_CLOUD_CONNECTIONS_TELE_H_
 
-#include "aether/tele.h"
+#include "aether/tele.h"  // IWYU pragma: exports
 
 AE_TELE_MODULE(kCloudClientConnection, 30, 110, 112);
 

aether/cloud_connections/cloud_server_connection.cpp

@@ -18,59 +18,49 @@
 
 #include "aether/server.h"
 
-#include "aether/cloud_connections/cloud_connections_tele.h"
-
 namespace ae {
+
 CloudServerConnection::CloudServerConnection(
     Ptr<Server> const& server, IServerConnectionFactory& connection_factory)
     : server_{server},
       connection_factory_{&connection_factory},
       priority_{},
-      is_connection_{},
       is_quarantined_{} {}
 
 std::size_t CloudServerConnection::priority() const { return priority_; }
-
-bool CloudServerConnection::quarantine() const { return is_quarantined_; }
-void CloudServerConnection::quarantine(bool value) { is_quarantined_ = value; }
+void CloudServerConnection::SetPriority(std::size_t priority) {
+  priority_ = priority;
+}
 
 void CloudServerConnection::Restream() {
-  if (client_connection_) {
+  if (client_connection_.get() != nullptr) {
     client_connection_->Restream();
   }
 }
 
-bool CloudServerConnection::BeginConnection(std::size_t priority) {
-  priority_ = priority;
-  AE_TELED_DEBUG("Begin connection server_id={}, priority={}, is_connection={}",
-                 server()->server_id, priority_, is_connection_);
-  if (!is_connection_) {
-    is_connection_ = true;
-    // Make new connection
-    client_connection_.Reset();
-    client_connection_ = connection_factory_->CreateConnection(server());
-    return true;
-  }
-  return false;
+bool CloudServerConnection::quarantine() const { return is_quarantined_; }
+void CloudServerConnection::SetQuarantine(bool value) {
+  is_quarantined_ = value;
 }
 
-void CloudServerConnection::EndConnection(std::size_t priority) {
-  priority_ = priority;
-  AE_TELED_DEBUG("End connection server_id={}, priority={}, is_connection={}",
-                 server()->server_id, priority_, is_connection_);
-  is_connection_ = false;
+bool CloudServerConnection::Connect() {
+  client_connection_.Reset();
+  client_connection_ = connection_factory_->CreateConnection(server());
+  return client_connection_.get() != nullptr;
 }
 
+void CloudServerConnection::Disconnect() { client_connection_.Reset(); }
+
 ClientServerConnection* CloudServerConnection::client_connection() {
-  if (client_connection_) {
+  if (client_connection_.get() != nullptr) {
     return client_connection_.get();
   }
   return nullptr;
 }
 
 Ptr<Server> CloudServerConnection::server() const {
   auto server = server_.Lock();
-  assert(server);
+  assert(server.get() != nullptr);
   return server;
 }
 

aether/cloud_connections/cloud_server_connection.h

@@ -17,55 +17,31 @@
 #ifndef AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTION_H_
 #define AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTION_H_
 
-#include "aether/ptr/rc_ptr.h"
-#include "aether/obj/obj_ptr.h"
 #include "aether/ptr/ptr_view.h"
+#include "aether/ptr/rc_ptr.h"
+
 #include "aether/server_connections/client_server_connection.h"
 #include "aether/server_connections/iserver_connection_factory.h"
 
 namespace ae {
 class Server;
-/**
- * \brief Represent the connection to the server.
- * It's not the actual transport but the all information about the server.
- */
+
 class CloudServerConnection {
  public:
   CloudServerConnection(Ptr<Server> const& server,
                         IServerConnectionFactory& connection_factory);
 
   std::size_t priority() const;
+  void SetPriority(std::size_t priority);
 
   void Restream();
 
-  /**
-   * \brief It's possible to put failed server to quarantine.
-   * Server in quarantine should not be used.
-   */
   bool quarantine() const;
-  void quarantine(bool value);
+  void SetQuarantine(bool value);
 
-  /**
-   * \brief Call to begin connection with specified priority.
-   * If connection already established, do nothing.
-   * If connection is not created, create it.
-   * \param priority the new priority value.
-   * Priority is used to sort connection in cloud for performing requests.
-   */
-  bool BeginConnection(std::size_t priority);
-  /**
-   * \brief Call to end connection with specified priority.
-   * \param priority the new priority value.
-   * Priority is used to sort connection in cloud for selecting the most
-   * prioritized servers.
-   */
-  void EndConnection(std::size_t priority);
+  bool Connect();
+  void Disconnect();
 
-  /**
-   * \brief The stream to write data for that server.
-   * Should be called after BeginConnection.
-   * \return The client-server connection object or nullptr.
-   */
   ClientServerConnection* client_connection();
 
   Ptr<Server> server() const;
@@ -75,7 +51,6 @@ class CloudServerConnection {
   IServerConnectionFactory* connection_factory_;
   RcPtr<ClientServerConnection> client_connection_;
   std::size_t priority_;
-  bool is_connection_;
   bool is_quarantined_;
 };
 }  // namespace ae