Fix lost notification and potential deadlock

Author: everoddandeven

src/cpp/daemon/py_monero_daemon.h

@@ -21,13 +21,17 @@ class PyMoneroDaemonListener : public monero_daemon_listener {
 
 class PyMoneroBlockNotifier : public PyMoneroDaemonListener {
 public:
-  boost::mutex* temp;
-  boost::condition_variable* cv;
-  PyMoneroBlockNotifier(boost::mutex* temp, boost::condition_variable* cv) { this->temp = temp; this->cv = cv; }
+  PyMoneroBlockNotifier(boost::mutex* temp, boost::condition_variable* cv, bool* ready) { this->temp = temp; this->cv = cv; this->ready = ready; }
   void on_block_header(const std::shared_ptr<monero::monero_block_header>& header) override {
+    boost::mutex::scoped_lock lock(*temp);
     m_last_header = header;
+    *ready = true;
     cv->notify_one();
   }
+private:
+  boost::mutex* temp;
+  boost::condition_variable* cv;
+  bool* ready;
 };
 
 class PyMoneroDaemon {

src/cpp/daemon/py_monero_daemon_rpc.cpp

@@ -616,19 +616,20 @@ void PyMoneroDaemonRpc::stop() {
 }
 
 std::shared_ptr<monero::monero_block_header> PyMoneroDaemonRpc::wait_for_next_block_header() {
-  // use mutex and condition variable to wait for block
+  // use mutex and condition variable with predicate to wait for block
   boost::mutex temp;
   boost::condition_variable cv;
+  bool ready = false;
 
   // create listener which notifies condition variable when block is added
-  auto block_listener = std::make_shared<PyMoneroBlockNotifier>(&temp, &cv);
+  auto block_listener = std::make_shared<PyMoneroBlockNotifier>(&temp, &cv, &ready);
 
   // register the listener
   add_listener(block_listener);
 
   // wait until condition variable is notified
   boost::mutex::scoped_lock lock(temp);
-  cv.wait(lock);
+  cv.wait(lock, [&]() { return ready; });
 
   // unregister the listener
   remove_listener(block_listener);

tests/test_monero_connection_manager.py

@@ -21,6 +21,7 @@ class TestMoneroConnectionManager:
     """Proxy used to simulate offline servers"""
 
     _cm: MoneroConnectionManager | None = None
+    """Connection manager test instance."""
 
     #region Fixtures
 
@@ -122,7 +123,7 @@ def test_connection_manager(self, connection_manager: MoneroConnectionManager, c
 
         # auto connect to the best available connection
         connection_manager.start_polling(Utils.SYNC_PERIOD_IN_MS)
-        listener.wait_for_change(Utils.SYNC_PERIOD_IN_MS, "Waiting for auto connect to best available connection")
+        listener.wait_for_change(num_expected_changes + 1, Utils.SYNC_PERIOD_IN_MS, f"Waiting for auto connect to best available connection")
         assert connection_manager.is_connected()
         connection = connection_manager.get_connection()
         assert connection is not None
@@ -166,7 +167,7 @@ def test_connection_manager(self, connection_manager: MoneroConnectionManager, c
                 continue
             conn.proxy_uri = self.OFFLINE_PROXY_URI
 
-        listener.wait_for_change(Utils.SYNC_PERIOD_IN_MS, "Simulating priotizized servers shut down")
+        listener.wait_for_change(num_expected_changes + 1, Utils.SYNC_PERIOD_IN_MS, "Simulating priotizized servers shut down")
         assert connection_manager.is_connected() is False, f"{connection_manager.get_connection().serialize()}"
         connection = connection_manager.get_connection()
 
@@ -330,7 +331,7 @@ def test_connection_manager(self, connection_manager: MoneroConnectionManager, c
             poll_type=MoneroConnectionPollType.CURRENT
         )
 
-        listener.wait_for_change(Utils.SYNC_PERIOD_IN_MS, "Polling current connection")
+        listener.wait_for_change(num_expected_changes + 1, Utils.SYNC_PERIOD_IN_MS, "Polling current connection")
         assert connection_manager.is_connected() is True
         num_expected_changes += 1
         assert num_expected_changes == listener.num_changed_connections
@@ -340,7 +341,7 @@ def test_connection_manager(self, connection_manager: MoneroConnectionManager, c
         num_expected_changes += 1
         assert num_expected_changes == listener.num_changed_connections
         connection_manager.start_polling(period_ms=Utils.SYNC_PERIOD_IN_MS, poll_type=MoneroConnectionPollType.ALL)
-        listener.wait_for_change(Utils.SYNC_PERIOD_IN_MS, "Polling all connections")
+        listener.wait_for_change(num_expected_changes + 1, Utils.SYNC_PERIOD_IN_MS, "Polling all connections")
         assert connection_manager.is_connected() is True
         num_expected_changes += 1
         assert num_expected_changes == listener.num_changed_connections
@@ -351,7 +352,7 @@ def test_connection_manager(self, connection_manager: MoneroConnectionManager, c
         for con in ordered_connections:
             con.proxy_uri = self.OFFLINE_PROXY_URI
 
-        listener.wait_for_change(Utils.SYNC_PERIOD_IN_MS, "Simulating total shut down")
+        listener.wait_for_change(num_expected_changes + 1, Utils.SYNC_PERIOD_IN_MS, "Simulating total shut down")
         assert connection.is_online() is False, f"Expected offline connection: {connection.serialize()}"
         num_expected_changes += 1
         assert num_expected_changes == listener.num_changed_connections

tests/utils/connection_change_collector.py

@@ -34,19 +34,21 @@ def on_connection_changed(self, connection: Optional[MoneroRpcConnection]) -> No
         logger.debug(f"Collecting connection change: {conn_str}")
         self.changed_connections.append(connection)
 
-    def wait_for_change(self, interval_ms: int = 5000, custom_message: str = "Waiting for connection change") -> Optional[MoneroRpcConnection]:
+    def wait_for_change(self, expected_num_changes: int, interval_ms: int = 5000, custom_message: str = "Waiting for connection change") -> Optional[MoneroRpcConnection]:
         """
         Wait until a connection change occurs.
 
+        :param int expected_num_changes: expected number of connection changes to wait for.
         :param int interval_ms: custom check interval in milliseconds (default 5000).
         :param str custom_message: custom message to show in debug during wait.
         :returns MoneroRpcConnection | None: changed connection.
         """
 
         last_num_changes: int = self.num_changed_connections
-        while self.num_changed_connections <= last_num_changes:
-            logger.debug(f"{custom_message} (connections {last_num_changes})...")
+        while expected_num_changes > last_num_changes:
+            logger.debug(f"{custom_message} (changes {last_num_changes}/{expected_num_changes})...")
             GenUtils.wait_for(interval_ms)
+            last_num_changes = self.num_changed_connections
 
         logger.debug(f"Connection changed (connections {self.num_changed_connections}).")