diff --git a/.opencode/agents/tester.md b/.opencode/agents/tester.md
index 980e53ac..bfd75818 100644
--- a/.opencode/agents/tester.md
+++ b/.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
diff --git a/aether/ae_actions/ae_actions_tele.h b/aether/ae_actions/ae_actions_tele.h
index f8a24a0c..7955467c 100644
--- a/aether/ae_actions/ae_actions_tele.h
+++ b/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);
diff --git a/aether/ae_actions/ping.cpp b/aether/ae_actions/ping.cpp
index 1a380d45..02f21238 100644
--- a/aether/ae_actions/ping.cpp
+++ b/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() {
diff --git a/aether/ae_actions/ping.h b/aether/ae_actions/ping.h
index 1be813ed..f7665a65 100644
--- a/aether/ae_actions/ping.h
+++ b/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
diff --git a/aether/client_messages/client_messages_tele.h b/aether/client_messages/client_messages_tele.h
index 4cf1c6c1..89da6c11 100644
--- a/aether/client_messages/client_messages_tele.h
+++ b/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);
 
diff --git a/aether/client_messages/p2p_message_stream_manager.cpp b/aether/client_messages/p2p_message_stream_manager.cpp
index aa9df6c1..caf7f15c 100644
--- a/aether/client_messages/p2p_message_stream_manager.cpp
+++ b/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);
 
diff --git a/aether/cloud_connections/cloud_connections_tele.h b/aether/cloud_connections/cloud_connections_tele.h
index 022b4969..0e1ad021 100644
--- a/aether/cloud_connections/cloud_connections_tele.h
+++ b/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);
 
diff --git a/aether/cloud_connections/cloud_server_connection.cpp b/aether/cloud_connections/cloud_server_connection.cpp
index be6193f2..422758b3 100644
--- a/aether/cloud_connections/cloud_server_connection.cpp
+++ b/aether/cloud_connections/cloud_server_connection.cpp
@@ -18,51 +18,41 @@
 
 #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;
@@ -70,7 +60,7 @@ ClientServerConnection* CloudServerConnection::client_connection() {
 
 Ptr<Server> CloudServerConnection::server() const {
   auto server = server_.Lock();
-  assert(server);
+  assert(server.get() != nullptr);
   return server;
 }
 
diff --git a/aether/cloud_connections/cloud_server_connection.h b/aether/cloud_connections/cloud_server_connection.h
index 563a386e..338583cc 100644
--- a/aether/cloud_connections/cloud_server_connection.h
+++ b/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
diff --git a/aether/cloud_connections/cloud_server_connections.cpp b/aether/cloud_connections/cloud_server_connections.cpp
index 4d7d32e7..e3273f68 100644
--- a/aether/cloud_connections/cloud_server_connections.cpp
+++ b/aether/cloud_connections/cloud_server_connections.cpp
@@ -13,19 +13,42 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 #include "aether/cloud_connections/cloud_server_connections.h"
 
 #include <algorithm>
+#include <cassert>
+#include <chrono>
+#include <cstdint>
+#include <utility>
 
 #include "aether/api_protocol/api_protocol.h"
 #include "aether/server.h"
 #include "aether/server_connections/server_connection.h"
 
-#include "aether/tele.h"
+#include "aether/cloud_connections/cloud_connections_tele.h"
 
 namespace ae {
 
+namespace {
+auto SelectedServersLog(
+    [[maybe_unused]] std::vector<CloudServerConnection*> const&
+        selected_servers) {
+#if DEBUG
+  std::vector<ServerId> server_ids;
+  server_ids.reserve(selected_servers.size());
+  for (auto* server_connection : selected_servers) {
+    server_ids.emplace_back(server_connection->server()->server_id);
+  }
+  return server_ids;
+#else
+  return "!not a debug!";
+#endif
+}
+}  // namespace
+
+static constexpr auto kCloudServerQuarantineTime =
+    std::chrono::milliseconds{AE_CLOUD_SERVER_QUARANTINE_TIME_MS};
+
 cloud_server_connections_internal::EmptyConnectionsWA::EmptyConnectionsWA(
     AeContext const& ae_context) {
   ae_context.scheduler().Task(
@@ -56,7 +79,6 @@ cloud_server_connections_internal::ReplicaWA::ReplicaWA(
         });
   }
 }
-
 void cloud_server_connections_internal::ReplicaWA::Stop() noexcept {
   for (auto* action : swas_) {
     action->Stop();
@@ -72,204 +94,209 @@ CloudServerConnections::CloudServerConnections(
       connection_factory_{std::move(connection_factory)},
       max_connections_{max_connections} {
   InitServerConnections();
-  InitServers();
+  ReconcileServers();
 }
 
 CloudServerConnections::ServersUpdate::Subscriber
 CloudServerConnections::servers_update_event() {
   return servers_update_event_;
 }
-
+CloudServerConnections::ServerQuarantineEvent::Subscriber
+CloudServerConnections::server_quarantined_event() {
+  return server_quarantined_event_;
+}
+CloudServerConnections::ServerQuarantineEvent::Subscriber
+CloudServerConnections::server_quarantine_release_event() {
+  return server_quarantine_release_event_;
+}
 std::vector<CloudServerConnection*> const&
 CloudServerConnections::selected_servers() const {
   return selected_servers_;
 }
-
 std::vector<CloudServerConnection*> const& CloudServerConnections::servers() {
   return all_servers_;
 }
-
 std::size_t CloudServerConnections::count_connections() const {
   return selected_servers_.size();
 }
-
 std::size_t CloudServerConnections::max_connections() const {
   return max_connections_;
 }
-
 void CloudServerConnections::Restream() {
-  // restream all servers
-  for (auto const& server : selected_servers_) {
+  for (auto* server : selected_servers_) {
     server->Restream();
   }
 }
 
 void CloudServerConnections::InitServerConnections() {
-  server_connections_.clear();
   auto cloud = cloud_.Lock();
-  assert(cloud);
+  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_);
   }
   all_servers_.clear();
   all_servers_.reserve(server_connections_.size());
-  for (auto& s : server_connections_) {
-    all_servers_.emplace_back(&s);
+  for (auto& server : server_connections_) {
+    all_servers_.emplace_back(&server);
   }
 }
 
-void CloudServerConnections::InitServers() {
-  AE_TELED_DEBUG("Init servers");
-  auto server_candidates = ServerCandidates();
-  // reselect servers if required
-  if (selected_servers_.size() >=
-      std::min(server_candidates.size(), max_connections_)) {
+void CloudServerConnections::SubscribeToServerState(
+    CloudServerConnection& server_connection) {
+  auto* conn = server_connection.client_connection();
+  if (conn == nullptr) {
     return;
   }
-  SelectServers(server_candidates);
+  auto const key = reinterpret_cast<std::uintptr_t>(&server_connection);
+  auto& subs = server_subs_[key] = {};
+  subs.state_sub = conn->stream_update_event().Subscribe(
+      [this, sc{&server_connection}, conn]() {
+        if (conn->stream_info().link_state == LinkState::kLinkError) {
+          QuarantineServer(*sc);
+          return true;
+        }
+        return false;
+      });
+  subs.error_sub = conn->server_connection().server_error_event().Subscribe(
+      [this, sc{&server_connection}]() { QuarantineServer(*sc); });
 }
-
-void CloudServerConnections::SelectServers(
-    std::vector<CloudServerConnection*> const& servers) {
-  auto select_count = std::min(servers.size(), max_connections_);
-
-  auto get_ids = [&]([[maybe_unused]] auto const& ss) noexcept {
-#if DEBUG
-    std::vector<ServerId> sids;
-    sids.reserve(ss.size());
-    for (auto const* server : ss) {
-      sids.emplace_back(server->server()->server_id);
-    }
-    return sids;
-#else
-    return "!not debug!";
-#endif
-  };
-
-  AE_TELED_DEBUG("Select servers count {} from sids [{}]", select_count,
-                 get_ids(servers));
-
-  selected_servers_.clear();
-  selected_servers_.reserve(select_count);
-
-  std::size_t i = 0;
-  // Select required count of servers.
-  for (; i < select_count; i++) {
-    auto* serv = servers[i];
-    selected_servers_.emplace_back(serv);
-    // update server priority and if it begins new connection update
-    // subscriptions
-    if (serv->BeginConnection(i)) {
-      SubscribeToServerState(*serv);
-    }
+void CloudServerConnections::UnsubscribeFromServerState(
+    CloudServerConnection& server_connection) {
+  auto const key = reinterpret_cast<std::uintptr_t>(&server_connection);
+  auto it = server_subs_.find(key);
+  if (it == server_subs_.end()) {
+    return;
   }
-  // Also explicitly end the other servers connection.
-  for (; i < servers.size(); i++) {
-    servers[i]->EndConnection(i);
+  it->second.state_sub.Reset();
+  it->second.error_sub.Reset();
+  if (!it->second.quarantine_sub) {
+    server_subs_.erase(it);
   }
-
-  AE_TELED_DEBUG("Selected servers [{}]", get_ids(selected_servers_));
-  servers_update_event_.Emit();
 }
 
-void CloudServerConnections::SubscribeToServerState(
+void CloudServerConnections::QuarantineServer(
     CloudServerConnection& server_connection) {
-  auto bad_server = [this, sc{&server_connection}]() {
-    // TODO: add the policy how to change the server priority on failure
-    // put server in quarantine and make it the least prioritized
-    auto new_priority = sc->priority() + server_connections_.size();
-    sc->EndConnection(new_priority);
-    QuarantineTimer(*sc);
-    UnselectServer(*sc);
-    ReselectServers();
-  };
-
-  auto* conn = server_connection.client_connection();
-  if (conn == nullptr) {
-    bad_server();
+  auto const key = reinterpret_cast<std::uintptr_t>(&server_connection);
+  if (server_connection.quarantine()) {
     return;
   }
+  AE_TELED_DEBUG("Quarantine server server_id={} priority={}",
+                 server_connection.server()->server_id,
+                 server_connection.priority());
+  UnsubscribeFromServerState(server_connection);
+  if (std::erase(selected_servers_, &server_connection) != 0) {
+    UpdateSelectedPriorities();
+    servers_update_event_.Emit();
+  }
+  server_connection.SetPriority(server_connections_.size());
+  server_connection.SetQuarantine(true);
+  server_quarantined_event_.Emit(&server_connection);
+  server_subs_[key].quarantine_sub = ae_context_.scheduler().DelayedTask(
+      [this, sc{&server_connection}, key]() {
+        ReleaseQuarantinedServer(*sc, key);
+      },
+      kCloudServerQuarantineTime);
+  if (!server_subs_[key].quarantine_sub) {
+    assert(false && "Failed to schedule quarantine release task");
+  }
+  ScheduleReconcileServers();
+}
 
-  auto if_bad_connection = [bad_server, conn,
-                            sid{server_connection.server()->server_id}]() {
-    AE_TELED_DEBUG("Server connection id {}, link state {}", sid,
-                   conn->stream_info().link_state);
-    if (conn->stream_info().link_state == LinkState::kLinkError) {
-      bad_server();
-      return true;
-    }
-    return false;
-  };
-
-  // any link error leads to reselect the servers
-  if (if_bad_connection()) {
+void CloudServerConnections::ReleaseQuarantinedServer(
+    CloudServerConnection& server_connection, std::uintptr_t key) {
+  if (!server_connection.quarantine()) {
     return;
   }
-  server_state_subs_[reinterpret_cast<std::uintptr_t>(&server_connection)] =
-      conn->stream_update_event().Subscribe(if_bad_connection);
+  AE_TELED_DEBUG("Release quarantined server server_id={} priority={}",
+                 server_connection.server()->server_id,
+                 server_connection.priority());
+  server_quarantine_release_event_.Emit(&server_connection);
+  server_connection.Disconnect();
+  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);
+    }
+  }
+  ScheduleReconcileServers();
 }
 
-void CloudServerConnections::ReselectServers() {
-  // reselct servers on next cycle
+void CloudServerConnections::ScheduleReconcileServers() {
   if (defer_sub_) {
     return;
   }
-  defer_sub_ = ae_context_.scheduler().Task([&]() {
+  defer_sub_ = ae_context_.scheduler().Task([this]() {
     defer_sub_.Reset();
-    InitServers();
+    ReconcileServers();
   });
 }
 
-void CloudServerConnections::UnselectServer(
-    CloudServerConnection& server_connection) {
-  auto old_size = selected_servers_.size();
-  std::erase(selected_servers_, &server_connection);
-  if (selected_servers_.size() < old_size) {
-    AE_TELED_DEBUG("Servers unselected, remaining count {}",
-                   selected_servers_.size());
+void CloudServerConnections::ReconcileServers() {
+  if (selected_servers_.size() >= max_connections_) {
+    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());
+  auto emplaced = false;
+  for (auto* candidate : candidates) {
+    if (selected_servers_.size() >= max_connections_) {
+      break;
+    }
+    candidate->SetPriority(selected_servers_.size());
+    candidate->Connect();
+    auto* conn = candidate->client_connection();
+    if (conn == nullptr ||
+        conn->stream_info().link_state == LinkState::kLinkError) {
+      AE_TELED_DEBUG(
+          "Candidate unusable during reconcile server_id={} priority={}",
+          candidate->server()->server_id, candidate->priority());
+      QuarantineServer(*candidate);
+      continue;
+    }
+    selected_servers_.emplace_back(candidate);
+    SubscribeToServerState(*candidate);
+    emplaced = true;
+  }
+  if (emplaced) {
+    AE_TELED_DEBUG("Selected servers reconciled selected_servers={}",
+                   SelectedServersLog(selected_servers_));
+    servers_update_event_.Emit();
   }
 }
 
-void CloudServerConnections::QuarantineTimer(
-    CloudServerConnection& server_connection) {
-  AE_TELED_DEBUG("Set server {} to quarantine",
-                 server_connection.server()->server_id);
-  static constexpr Duration kQuarantineDuration =
-      std::chrono::milliseconds{AE_CLOUD_SERVER_QUARANTINE_TIME_MS};
-  // TODO: add task subscription here
-  ae_context_.scheduler().DelayedTask(
-      [this, sc{&server_connection}]() {
-        AE_TELED_DEBUG("Release from quarantine server {}",
-                       sc->server()->server_id);
-        sc->quarantine(false);
-        // update selected servers
-        ReselectServers();
-      },
-      kQuarantineDuration);
-  server_connection.quarantine(true);
+bool CloudServerConnections::IsSelected(CloudServerConnection* sc) const {
+  return std::find(selected_servers_.begin(), selected_servers_.end(), sc) !=
+         selected_servers_.end();
+}
+
+void CloudServerConnections::UpdateSelectedPriorities() {
+  for (std::size_t i = 0; i < selected_servers_.size(); ++i) {
+    selected_servers_.at(i)->SetPriority(i);
+  }
 }
 
-std::vector<CloudServerConnection*> CloudServerConnections::ServerCandidates() {
+std::vector<CloudServerConnection*>
+CloudServerConnections::ReplacementCandidates() {
   std::vector<CloudServerConnection*> servers;
   servers.reserve(server_connections_.size());
   for (auto& s : server_connections_) {
-    if (s.quarantine()) {
-      continue;
+    if (!s.quarantine() && !IsSelected(&s)) {
+      servers.emplace_back(&s);
     }
-    servers.emplace_back(&s);
   }
-
-  // sort list of servers by it's priorities
-  std::sort(std::begin(servers), std::end(servers),
+  std::sort(servers.begin(), servers.end(),
             [](auto const* left, auto const* right) {
-              // TODO: add sorting by external policy
-              // Sort by priority
-              // The default priority is 0
-              // but it's changed lately after servers selected
               return left->priority() < right->priority();
             });
-
   return servers;
 }
 
@@ -279,12 +306,11 @@ WriteAction& CloudServerConnections::CallApi(ApiCall const& api_caller,
   ForServers(
       [&](CloudServerConnection* sc) {
         auto* conn = sc->client_connection();
-        assert((conn != nullptr) && "Client connection is null");
+        assert(conn != nullptr);
         swas.emplace_back(&conn->AuthorizedApiCall(
             SubApi<AuthorizedApi>{[&](auto& api) { api_caller(api, sc); }}));
       },
       policy);
-
   if (swas.empty()) {
     return EmptyWriteAction();
   }
@@ -293,17 +319,14 @@ WriteAction& CloudServerConnections::CallApi(ApiCall const& api_caller,
   }
   return ReplicaWriteAction(std::move(swas));
 }
-
 WriteAction& CloudServerConnections::EmptyWriteAction() {
   if (!empty_wa_ || empty_wa_->is_finished()) {
     empty_wa_.emplace(ae_context_);
   }
   return *empty_wa_;
 }
-
 WriteAction& CloudServerConnections::ReplicaWriteAction(
     std::vector<WriteAction*>&& swas) {
-  // FIXME: this vector only grows
   return replica_was_.emplace_back(std::move(swas));
 }
 
diff --git a/aether/cloud_connections/cloud_server_connections.h b/aether/cloud_connections/cloud_server_connections.h
index 4ba70924..ca04a82b 100644
--- a/aether/cloud_connections/cloud_server_connections.h
+++ b/aether/cloud_connections/cloud_server_connections.h
@@ -17,20 +17,21 @@
 #define AETHER_CLOUD_CONNECTIONS_CLOUD_SERVER_CONNECTIONS_H_
 
 #include <map>
-#include <vector>
 #include <memory>
 #include <optional>
+#include <vector>
 
-#include "aether/cloud.h"
-#include "aether/ptr/ptr.h"
-#include "aether/ptr/ptr_view.h"
 #include "aether/ae_context.h"
 #include "aether/events/events.h"
 #include "aether/events/multi_subscription.h"
+#include "aether/ptr/ptr.h"
+#include "aether/ptr/ptr_view.h"
 #include "aether/write_action/write_action.h"
-#include "aether/cloud_connections/request_policy.h"
+
+#include "aether/cloud.h"
 #include "aether/cloud_connections/cloud_callbacks.h"
 #include "aether/cloud_connections/cloud_server_connection.h"
+#include "aether/cloud_connections/request_policy.h"
 #include "aether/server_connections/iserver_connection_factory.h"
 
 namespace ae {
@@ -57,6 +58,7 @@ class ReplicaWA final : public WriteAction {
 class CloudServerConnections {
  public:
   using ServersUpdate = Event<void()>;
+  using ServerQuarantineEvent = Event<void(CloudServerConnection*)>;
 
   CloudServerConnections(
       AeContext const& ae_context, Ptr<Cloud> const& cloud,
@@ -67,6 +69,8 @@ class CloudServerConnections {
    * \brief The event then top list of the servers were updated.
    */
   ServersUpdate::Subscriber servers_update_event();
+  ServerQuarantineEvent::Subscriber server_quarantined_event();
+  ServerQuarantineEvent::Subscriber server_quarantine_release_event();
   /**
    * \brief List of currently selected servers in priority order
    */
@@ -140,15 +144,24 @@ class CloudServerConnections {
   WriteAction& ReplicaWriteAction(std::vector<WriteAction*>&& swas);
 
   void InitServerConnections();
-  void InitServers();
-  void SelectServers(std::vector<CloudServerConnection*> const& servers);
+  // Caller must unsubscribe before re-subscribing the same server.
   void SubscribeToServerState(CloudServerConnection& server_connection);
-  void ReselectServers();
-
-  void UnselectServer(CloudServerConnection& server_connection);
-  void QuarantineTimer(CloudServerConnection& server_connection);
-
-  std::vector<CloudServerConnection*> ServerCandidates();
+  void UnsubscribeFromServerState(CloudServerConnection& server_connection);
+  void QuarantineServer(CloudServerConnection& server_connection);
+  void ReleaseQuarantinedServer(CloudServerConnection& server_connection,
+                                std::uintptr_t key);
+  void ScheduleReconcileServers();
+  void ReconcileServers();
+
+  bool IsSelected(CloudServerConnection* sc) const;
+  void UpdateSelectedPriorities();
+  std::vector<CloudServerConnection*> ReplacementCandidates();
+
+  struct ServerSubscriptions {
+    Subscription state_sub;
+    Subscription error_sub;
+    TaskSubscription quarantine_sub;
+  };
 
   AeContext ae_context_;
   PtrView<Cloud> cloud_;
@@ -161,8 +174,10 @@ class CloudServerConnections {
   std::vector<CloudServerConnection*> selected_servers_;
 
   ServersUpdate servers_update_event_;
+  ServerQuarantineEvent server_quarantined_event_;
+  ServerQuarantineEvent server_quarantine_release_event_;
 
-  std::map<std::uintptr_t, Subscription> server_state_subs_;
+  std::map<std::uintptr_t, ServerSubscriptions> server_subs_;
   TaskSubscription defer_sub_;
 
   std::optional<cloud_server_connections_internal::EmptyConnectionsWA>
diff --git a/aether/cloud_connections/cloud_subscription.cpp b/aether/cloud_connections/cloud_subscription.cpp
index 6ef1797c..0112357a 100644
--- a/aether/cloud_connections/cloud_subscription.cpp
+++ b/aether/cloud_connections/cloud_subscription.cpp
@@ -55,13 +55,13 @@ CloudEventListener& CloudEventListener::operator=(
 }
 
 void CloudEventListener::ServersUpdate() {
-  // clean old subscriptions and make new
   subscriptions_.Reset();
   cloud_connection_->ForServers(
       [&](auto* sc) {
         auto* conn = sc->client_connection();
         assert((conn != nullptr) && "ClientConnection is null");
-        subscriptions_ += subscriber_(conn->client_safe_api(), sc);
+        auto sub = subscriber_(conn->client_safe_api(), sc);
+        subscriptions_ += std::move(sub);
       },
       request_policy_);
 }
diff --git a/aether/cloud_connections/ping_cloud_servers.cpp b/aether/cloud_connections/ping_cloud_servers.cpp
index b27676b7..9aba9991 100644
--- a/aether/cloud_connections/ping_cloud_servers.cpp
+++ b/aether/cloud_connections/ping_cloud_servers.cpp
@@ -17,7 +17,8 @@
 #include "aether/cloud_connections/ping_cloud_servers.h"
 
 #include <cassert>
-#include <set>
+#include <type_traits>
+#include <variant>
 
 #if AE_ENABLE_PING
 
@@ -46,7 +47,7 @@ PingCloudServers::ServerPing::ServerPing(AeContext const& ae_context,
   // if it's to early for next rx wait a bit
   if ((timings.next_rx_point != TimePoint{}) &&
       (Now() < timings.next_rx_point)) {
-    AE_TELED_DEBUG("Wait a bit for next rx point till {:%H:%M:%S}",
+    AE_TELED_DEBUG("Wait a bit for next rx point till {}",
                    timings.next_rx_point);
     start_sub_ = ae_context_.scheduler().DelayedTask([&]() { Start(); },
                                                      timings.next_rx_point);
@@ -62,9 +63,11 @@ PingCloudServers::ServerPing::~ServerPing() = default;
 void PingCloudServers::ServerPing::Stop() {
   stop_ = true;
 
+  waiter_.reset();
   start_sub_.Reset();
   rx_window_sub_.Reset();
   restream_sub_.Reset();
+  link_state_sub_.Reset();
 
   ping_blocker_.Reset();
   rx_window_blocker_.Reset();
@@ -121,10 +124,11 @@ auto PingCloudServers::ServerPing::MakePing() {
                         timing_conf_.rx_window, c->ResponseTimeout());
 
           ping_blocker_ = policy_->AcquireSuspendBlock();
-          ping_->result_event().Subscribe([this](auto const& res) noexcept {
-            OnPingResult(res);
-            ping_blocker_.Reset();
-          });
+          ping_->result_event().Subscribe(
+              [this](Ping::PingResult const& res) noexcept {
+                OnPingResult(res);
+                ping_blocker_.Reset();
+              });
 
           // run ping request and open rx window
           auto const current_time = Now();
@@ -132,9 +136,8 @@ auto PingCloudServers::ServerPing::MakePing() {
           OpenRxWindow(current_time);
           next_ping_time_ = current_time + timing_conf_.interval;
           policy_->ReportNextServiceTime(priority_, next_ping_time_);
-          AE_TELED_DEBUG(
-              "Next ping time for priority {} at {:%H:%M:%S} after {:%S}",
-              priority_, next_ping_time_, timing_conf_.interval);
+          AE_TELED_DEBUG("Next ping time for priority {} at {} after {}",
+                         priority_, next_ping_time_, timing_conf_.interval);
 
           return ex::set_value(std::move(ctx.receiver));
         });
@@ -142,6 +145,9 @@ auto PingCloudServers::ServerPing::MakePing() {
 }
 
 void PingCloudServers::ServerPing::Start() {
+  if (stop_) {
+    return;
+  }
   waiter_.emplace(
       ae_context_,
       EnsureLinked() |
@@ -179,25 +185,33 @@ void PingCloudServers::ServerPing::Start() {
       });
 }
 
-void PingCloudServers::ServerPing::OnPingResult(
-    Result<Duration, int> const& res) {
+void PingCloudServers::ServerPing::OnPingResult(Ping::PingResult const& res) {
   auto* cc = cloud_sc_->client_connection();
   if (cc == nullptr) {
     AE_TELED_ERROR("Client connection is null");
     return;
   }
 
-  if (res) {
-    auto c = cc->server_connection().current_channel();
-    if (!c) {
-      AE_TELED_ERROR("Ping's channel value invalid");
-    } else {
-      c->channel_statistics().AddResponseTime(res.value());
-    }
-  } else {
-    AE_TELED_ERROR("Ping error!");
-    ScheduleRestream();
+  auto c = cc->server_connection().current_channel();
+  if (!c) {
+    AE_TELED_ERROR("Connection channel is null");
+    return;
   }
+
+  std::visit(
+      [this, c](auto const& value) {
+        using T = std::decay_t<decltype(value)>;
+        if constexpr (std::is_same_v<T, Ok<Duration>>) {
+          c->channel_statistics().AddResponseTime(value.value);
+        } else if constexpr (std::is_same_v<T, Ping::LateDuration>) {
+          AE_TELED_DEBUG("Got late ping duration");
+          c->channel_statistics().AddResponseTime(value.duration);
+        } else {
+          AE_TELED_ERROR("Ping error!");
+          ScheduleRestream();
+        }
+      },
+      res);
 }
 
 void PingCloudServers::ServerPing::OpenRxWindow(TimePoint sent_time) {
@@ -235,6 +249,12 @@ PingCloudServers::PingCloudServers(
           cloud_server_connections_->servers_update_event().Subscribe(
               MethodPtr<&PingCloudServers::ServersUpdate>{this})} {
   AE_TELED_INFO("PingCloudServers created");
+  server_quarantined_sub_ =
+      cloud_server_connections_->server_quarantined_event().Subscribe(
+          MethodPtr<&PingCloudServers::ServerQuarantined>{this});
+  server_quarantine_released_sub_ =
+      cloud_server_connections_->server_quarantine_release_event().Subscribe(
+          MethodPtr<&PingCloudServers::ServerQuarantineReleased>{this});
   ServersUpdate();
 }
 
@@ -243,37 +263,28 @@ PingCloudServers::~PingCloudServers() { task_sub_.Reset(); }
 void PingCloudServers::ServersUpdate() {
   AE_TELED_DEBUG("Servers update");
   if (!task_sub_) {
-    AE_TELED_DEBUG("Enqueue dispatch servers");
     auto blocker = policy_->AcquireSuspendBlock();
     task_sub_ = ae_context_.scheduler().Task(
         [this, blocker = std::move(blocker)]() mutable {
+          task_sub_.Reset();
           DispatchToServers();
         });
   }
 }
 
 void PingCloudServers::DispatchToServers() {
-  std::set<ServerId> visited_ids;
   cloud_server_connections_->ForServers(
-      [this, &visited_ids](CloudServerConnection* cloud_sc) {
+      [this](CloudServerConnection* cloud_sc) {
         if (cloud_sc == nullptr) {
           AE_TELED_ERROR("Visit empty cloud server connection!");
           return;
         }
         auto server = cloud_sc->server();
         if (server) {
-          visited_ids.insert(server->server_id);
           ReconcileServer(server, *cloud_sc);
         }
       },
       policy_->rx_targets());
-
-  // stop pings for servers not in connection anymore
-  for (auto& [sid, sp] : server_pings_) {
-    if (!visited_ids.contains(sid)) {
-      sp->Stop();
-    }
-  }
 }
 
 void PingCloudServers::ReconcileServer(Ptr<Server> const& server,
@@ -282,13 +293,46 @@ void PingCloudServers::ReconcileServer(Ptr<Server> const& server,
   auto const priority = cloud_sc.priority();
 
   auto it = server_pings_.find(server_id);
-  if ((it == server_pings_.end()) || (it->second->priority() != priority)) {
+  if ((it == server_pings_.end()) || (it->second->priority() != priority) ||
+      it->second->stopped()) {
+    if (it != server_pings_.end()) {
+      it->second.reset();
+    }
     server_pings_.insert_or_assign(
         server_id, std::make_unique<ServerPing>(ae_context_, *policy_, cloud_sc,
                                                 priority));
     return;
   }
 }
+
+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);
+  if (it != server_pings_.end()) {
+    it->second->Stop();
+  }
+}
+
+void PingCloudServers::ServerQuarantineReleased(
+    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);
+  if (it != server_pings_.end()) {
+    server_pings_.erase(it);
+  }
+}
 }  // namespace ae
 
 #endif
diff --git a/aether/cloud_connections/ping_cloud_servers.h b/aether/cloud_connections/ping_cloud_servers.h
index 97b19970..215087e3 100644
--- a/aether/cloud_connections/ping_cloud_servers.h
+++ b/aether/cloud_connections/ping_cloud_servers.h
@@ -50,6 +50,7 @@ class PingCloudServers {
     TimePoint next_service_time() const noexcept { return next_ping_time_; }
     std::size_t priority() const noexcept { return priority_; }
     RxTimingConf const& timing() const noexcept { return timing_conf_; }
+    bool stopped() const noexcept { return stop_; }
 
    private:
     void Start();
@@ -60,7 +61,7 @@ class PingCloudServers {
     auto EnsureLinked();
     auto MakePing();
 
-    void OnPingResult(Result<Duration, int> const& res);
+    void OnPingResult(Ping::PingResult const& res);
     void OpenRxWindow(TimePoint sent_time);
     void ScheduleRestream();
 
@@ -95,12 +96,16 @@ class PingCloudServers {
   void DispatchToServers();
   void ReconcileServer(Ptr<Server> const& server,
                        CloudServerConnection& cloud_sc);
+  void ServerQuarantined(CloudServerConnection* cloud_sc);
+  void ServerQuarantineReleased(CloudServerConnection* cloud_sc);
 
   AeContext ae_context_;
   CloudServerConnections* cloud_server_connections_;
   ClientConnectivityPolicy* policy_;
 
   Subscription servers_update_;
+  Subscription server_quarantined_sub_;
+  Subscription server_quarantine_released_sub_;
   TaskSubscription task_sub_;
 
   std::map<ServerId, std::unique_ptr<ServerPing>> server_pings_;
diff --git a/aether/work_cloud_api/client_api/client_api_safe.cpp b/aether/work_cloud_api/client_api/client_api_safe.cpp
index 5f36082d..2b937bd3 100644
--- a/aether/work_cloud_api/client_api/client_api_safe.cpp
+++ b/aether/work_cloud_api/client_api/client_api_safe.cpp
@@ -41,7 +41,6 @@ void ClientApiSafe::NewChildren([[maybe_unused]] std::vector<Uid> const& uids) {
 
 void ClientApiSafe::SendMessages(std::vector<AeMessage> const& messages) {
   for (auto const& msg : messages) {
-    AE_TELED_DEBUG("Received message uid:{}", msg.uid);
     send_message_event_.Emit(msg);
   }
 }
@@ -111,7 +110,6 @@ void ClientApiSafe::SendAccessCheckResults(
 }
 
 void ClientApiSafe::SendMessage(AeMessage const& msg) {
-  AE_TELED_DEBUG("Received message uid:{}", msg.uid);
   send_message_event_.Emit(msg);
 }