Use monotonic clock for timeouts; harden db-sync progress wait

cardano_node_tests/tests/test_blocks.py

@@ -375,19 +375,19 @@ def _save_state(curr_epoch: int) -> None:
         tip = cluster.g_query.get_tip()
         epoch_end = cluster.time_to_epoch_end(tip)
         curr_epoch = cluster.g_query.get_epoch(tip=tip)
-        curr_time = time.time()
-        epoch_end_timestamp = curr_time + epoch_end
-        test_end_timestamp = epoch_end_timestamp + (num_epochs * cluster.epoch_length_sec)
+        curr_time = time.monotonic()
+        epoch_end_deadline = curr_time + epoch_end
+        test_end_deadline = epoch_end_deadline + (num_epochs * cluster.epoch_length_sec)
 
         LOGGER.info(f"Checking blocks for {num_epochs} epochs.")
-        while curr_time < test_end_timestamp:
-            epoch_end = epoch_end_timestamp - curr_time
+        while curr_time < test_end_deadline:
+            epoch_end = epoch_end_deadline - curr_time
             if epoch_end < 15:
                 LOGGER.info(f"End of epoch {curr_epoch}, saving data.")
                 _save_state(curr_epoch)
 
                 curr_epoch = cluster.wait_for_new_epoch(padding_seconds=5)
-                epoch_end_timestamp = cluster.time_to_epoch_end() + time.time()
+                epoch_end_deadline = cluster.time_to_epoch_end() + time.monotonic()
 
             # Send tx
             src_addr, dst_addr = random.sample(payment_addrs, 2)
@@ -396,13 +396,13 @@ def _save_state(curr_epoch: int) -> None:
 
             cluster.g_transaction.send_tx(
                 src_address=src_addr.address,
-                tx_name=f"{temp_template}_{int(curr_time)}",
+                tx_name=f"{temp_template}_{int(time.time())}",
                 txouts=txouts,
                 tx_files=tx_files,
             )
 
             time.sleep(2)
-            curr_time = time.time()
+            curr_time = time.monotonic()
 
         # Save also data for the last epoch
         _save_state(cluster.g_query.get_epoch())
@@ -537,14 +537,14 @@ def _save_state(curr_epoch: int) -> dict[str, int]:
         )
 
         epoch_end = cluster.time_to_epoch_end(tip)
-        curr_time = time.time()
-        epoch_end_timestamp = curr_time + epoch_end
-        test_end_timestamp = epoch_end_timestamp + (num_epochs * cluster.epoch_length_sec)
+        curr_time = time.monotonic()
+        epoch_end_deadline = curr_time + epoch_end
+        test_end_deadline = epoch_end_deadline + (num_epochs * cluster.epoch_length_sec)
 
         blocks_db = {}
         LOGGER.info(f"Checking blocks for {num_epochs} epochs.")
-        while curr_time < test_end_timestamp:
-            epoch_end = epoch_end_timestamp - curr_time
+        while curr_time < test_end_deadline:
+            epoch_end = epoch_end_deadline - curr_time
             if epoch_end < 15:
                 LOGGER.info(f"End of epoch {curr_epoch}, saving data.")
                 blocks_db[curr_epoch] = _save_state(curr_epoch)
@@ -558,7 +558,7 @@ def _save_state(curr_epoch: int) -> dict[str, int]:
                     )
 
                 curr_epoch = cluster.wait_for_new_epoch(padding_seconds=5)
-                epoch_end_timestamp = cluster.time_to_epoch_end() + time.time()
+                epoch_end_deadline = cluster.time_to_epoch_end() + time.monotonic()
 
                 # Replace the node
                 if curr_epoch == reconf_epoch + 1:
@@ -595,13 +595,13 @@ def _save_state(curr_epoch: int) -> dict[str, int]:
 
             cluster.g_transaction.send_tx(
                 src_address=src_addr.address,
-                tx_name=f"{temp_template}_{int(curr_time)}",
+                tx_name=f"{temp_template}_{int(time.time())}",
                 txouts=txouts,
                 tx_files=tx_files,
             )
 
             time.sleep(2)
-            curr_time = time.time()
+            curr_time = time.monotonic()
 
         # Save also data for the last epoch
         curr_epoch = cluster.g_query.get_epoch()

cardano_node_tests/tests/test_tx_many_utxos.py

@@ -85,7 +85,7 @@ def many_utxos(
         temp_template = common.get_test_id(cluster)
 
         LOGGER.info("Generating lot of UTxO addresses, it will take a while.")
-        start = time.time()
+        start = time.monotonic()
         payment_addr = payment_addrs[0]
         out_addrs1 = [payment_addrs[1] for __ in range(200)]
         out_addrs2 = [payment_addrs[2] for __ in range(200)]
@@ -137,7 +137,7 @@ def many_utxos(
             out_addrs=out_addrs2,
             amount=10_000_000,
         )
-        end = time.time()
+        end = time.monotonic()
 
         retval = payment_addrs[1], payment_addrs[2]
 

cardano_node_tests/utils/dbsync_queries.py

@@ -1639,12 +1639,11 @@ def delete_reserved_pool_tickers() -> int:
         return affected_rows
 
 
-def query_db_sync_progress() -> float:
+def query_db_sync_progress() -> float | None:
     """Calculate blockchain sync percentage (0-100).
 
     Returns:
-        float: Sync percentage (0-100) if blocks exist
-        None: If no blocks in database
+        Sync percentage (0-100), or `None` when there are no blocks in the database.
     """
     query = (
         "SELECT"
@@ -1658,7 +1657,10 @@ def query_db_sync_progress() -> float:
 
     with execute(query=query) as cur:
         result = cur.fetchone()
-        return min(100.0, float(result[0])) if result else 0.0
+        # On an empty `block` table the aggregate query returns a single `(None,)` row
+        if result is None or result[0] is None:
+            return None
+        return min(100.0, float(result[0]))
 
 
 def query_rows_count(

cardano_node_tests/utils/dbsync_utils.py

@@ -512,7 +512,7 @@ def retry_query(*, query_func: tp.Callable, timeout: int = 20) -> tp.Any:
     The query is repeated until the expected data is returned, or `DbSyncTimeoutError` is
     raised when the timeout is reached.
     """
-    end_time = time.time() + timeout
+    end_time = time.monotonic() + timeout
     repeat = 0
 
     while True:
@@ -524,7 +524,7 @@ def retry_query(*, query_func: tp.Callable, timeout: int = 20) -> tp.Any:
             response = query_func()
             break
         except DbSyncNoResponseError as exc:
-            if time.time() < end_time:
+            if time.monotonic() < end_time:
                 repeat += 1
                 continue
             raise DbSyncTimeoutError(str(exc)) from exc
@@ -1751,24 +1751,25 @@ def check_off_chain_vote_fetch_error(*, voting_anchor_id: int) -> None:
 def wait_for_db_sync_completion(
     *, expected_progress: float = 99.0, timeout: int = 360, polling_interval: int = 5
 ) -> float:
-    """Wait for db-sync to reach at least 99% sync completion.
+    """Wait for db-sync to reach the expected sync completion.
 
     Args:
-        expected_progress: Expected completion as perctentage, 99% by default
-        timeout: Maximum time to wait in seconds
+        expected_progress: Expected completion as percentage, 99% by default
+        timeout: Maximum total time to wait in seconds
         polling_interval: Loop polling time in seconds
 
     Returns:
-        Final sync percentage achieved (>= 99)
+        Final sync percentage achieved (>= `expected_progress`)
 
     Raises:
-        TimeoutError: If sync doesn't reach 99% within timeout
+        DbSyncTimeoutError: If sync doesn't reach the expected progress within timeout
     """
-    start_time = time.time()
+    start_time = time.monotonic()
+    deadline = start_time + timeout
 
     def _query_func() -> float:
         dbsync_progress = dbsync_queries.query_db_sync_progress()
-        if not dbsync_progress:
+        if dbsync_progress is None:
             msg = "no result for query_db_sync_progress"
             raise DbSyncNoResponseError(msg)
         return dbsync_progress
@@ -1777,12 +1778,21 @@ def _query_func() -> float:
 
     # Poll until sync completes
     while dbsync_progress < expected_progress:
-        if time.time() - start_time > timeout:
-            err_msg = f"db-sync only reached {dbsync_progress}% after {timeout} seconds"
-            raise TimeoutError(err_msg)
+        if time.monotonic() > deadline:
+            elapsed = round(time.monotonic() - start_time)
+            err_msg = (
+                f"db-sync only reached {dbsync_progress:.2f}% "
+                f"(expected >= {expected_progress:.2f}%) "
+                f"after {elapsed} seconds"
+            )
+            raise DbSyncTimeoutError(err_msg)
         time.sleep(polling_interval)
-        dbsync_progress = dbsync_queries.query_db_sync_progress()
-        LOGGER.info(f"Progress of db-sync: {dbsync_queries.query_db_sync_progress():.2f}%")
+        # A `None` progress means there are no blocks in the database yet. Keep polling
+        # with the last known progress until the deadline is reached.
+        new_progress = dbsync_queries.query_db_sync_progress()
+        if new_progress is not None:
+            dbsync_progress = new_progress
+        LOGGER.info(f"Progress of db-sync: {dbsync_progress:.2f}%")
 
     return dbsync_progress