Refactor test suite to support starting components asynchronously

src/uhs/atomizer/archiver/controller.cpp

@@ -103,10 +103,12 @@ namespace cbdc::archiver {
 
     auto controller::init_atomizer_connection() -> bool {
         m_atomizer_network.cluster_connect(m_opts.m_atomizer_endpoints, false);
-        if(!m_atomizer_network.connected_to_one()) {
-            m_logger->error("Failed to connect to any atomizers.");
-            return false;
+        while(!m_atomizer_network.connected_to_one()) {
+            // TODO:  should we limit the number of attempts?
+            m_logger->warn("Waiting to connect to atomizers...");
+            std::this_thread::sleep_for(std::chrono::seconds(1));
         }
+        m_logger->info("Connected to atomizer network.");
 
         m_atomizer_handler_thread
             = m_atomizer_network.start_handler([&](auto&& pkt) {

src/uhs/atomizer/atomizer/controller.cpp

@@ -58,11 +58,15 @@ namespace cbdc::atomizer {
     }
 
     auto controller::init() -> bool {
-        if(!m_watchtower_network.cluster_connect(
-               m_opts.m_watchtower_internal_endpoints)) {
-            m_logger->error("Failed to connect to watchtowers.");
-            return false;
+        m_watchtower_network.cluster_connect(
+            m_opts.m_watchtower_internal_endpoints,
+            false);
+        while(!m_watchtower_network.connected_to_one()) {
+            // TODO:  should we limit the number of attempts?
+            m_logger->warn("Waiting to connect to watchtowers...");
+            std::this_thread::sleep_for(std::chrono::seconds(1));
         }
+        m_logger->info("Connected to watchtower network.");
 
         auto raft_params = nuraft::raft_params();
         raft_params.election_timeout_lower_bound_

src/uhs/atomizer/sentinel/controller.cpp

@@ -39,16 +39,20 @@ namespace cbdc::sentinel {
                            shard.second,
                            "...");
             auto sock = std::make_unique<network::tcp_socket>();
-            if(!sock->connect(shard)) {
-                m_logger->error("failed to connect");
-                return false;
+            while(!sock->connect(shard)) {
+                m_logger->warn("Failed to connect to",
+                               shard.first,
+                               ":",
+                               shard.second,
+                               ".  Retrying...");
+                std::this_thread::sleep_for(std::chrono::seconds(1));
             }
 
             const auto peer_id = m_shard_network.add(std::move(sock));
             const auto& shard_range = m_opts.m_shard_ranges[i];
             m_shard_data.push_back(shard_info{shard_range, peer_id});
 
-            m_logger->info("done");
+            m_logger->info("Done");
         }
 
         m_shard_dist = decltype(m_shard_dist)(0, m_shard_data.size() - 1);

src/uhs/atomizer/shard/controller.cpp

@@ -49,22 +49,30 @@ namespace cbdc::shard {
             return false;
         }
 
-        if(!m_archiver_client.init()) {
-            m_logger->error("Failed to connect to archiver");
-            return false;
+        static constexpr auto retry_delay = std::chrono::seconds(1);
+        while(!m_archiver_client.init()) {
+            m_logger->warn("Failed to connect to archiver.  Retrying...");
+            std::this_thread::sleep_for(retry_delay);
         }
-
-        if(!m_watchtower_network.cluster_connect(
-               m_opts.m_watchtower_internal_endpoints)) {
-            m_logger->error("Failed to connect to watchtowers.");
-            return false;
+        m_logger->info("Connected to archiver.");
+
+        m_watchtower_network.cluster_connect(
+            m_opts.m_watchtower_internal_endpoints,
+            false);
+        while(!m_watchtower_network.connected_to_one()) {
+            // TODO:  should we limit the number of attempts?
+            m_logger->warn("Waiting to connect to watchtowers...");
+            std::this_thread::sleep_for(retry_delay);
         }
+        m_logger->info("Connected to watchtower network.");
 
         m_atomizer_network.cluster_connect(m_opts.m_atomizer_endpoints, false);
-        if(!m_atomizer_network.connected_to_one()) {
-            m_logger->error("Failed to connect to any atomizers");
-            return false;
+        while(!m_atomizer_network.connected_to_one()) {
+            // TODO:  should we limit the number of attempts?
+            m_logger->warn("Waiting to connect to atomizers ...");
+            std::this_thread::sleep_for(retry_delay);
         }
+        m_logger->info("Connected to atomizer network.");
 
         m_atomizer_client = m_atomizer_network.start_handler([&](auto&& pkt) {
             return atomizer_handler(std::forward<decltype(pkt)>(pkt));

src/uhs/atomizer/watchtower/controller.cpp

@@ -73,14 +73,15 @@ auto cbdc::watchtower::controller::init() -> bool {
         // Since atomizers require a watchtower and the archiver requires an
         // atomizer, this has to be allowed to fail. The network will reconnect
         // when an atomizer comes online.
-        m_logger->warn("Failed to connect to any atomizers, waiting...");
+        m_logger->warn("Waiting to connect to atomizers...");
         std::this_thread::sleep_for(retry_delay);
     }
 
     while(!m_archiver_client.init()) {
-        m_logger->warn("Failed to connect to archiver, retrying...");
+        m_logger->warn("Failed to connect to archiver.  Retrying...");
         std::this_thread::sleep_for(retry_delay);
     }
+    m_logger->info("Connected to archiver.");
 
     m_atomizer_thread = m_atomizer_network.start_handler([&](auto&& pkt) {
         return atomizer_handler(std::forward<decltype(pkt)>(pkt));

tests/integration/atomizer_end_to_end_test.cpp

@@ -47,7 +47,7 @@ class atomizer_end_to_end_test : public ::testing::Test {
             ASSERT_TRUE(m_ctl_watchtower->init());
         });
 
-        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        // std::this_thread::sleep_for(std::chrono::milliseconds(100));
 
         ASSERT_TRUE(m_ctl_atomizer->init());
         ASSERT_TRUE(m_ctl_archiver->init());