Conform to rust api (#457)

Co-authored-by: Jean Demeusy <[email protected]>
Co-authored-by: ausias-armesto <[email protected]>
Co-authored-by: Tibor <[email protected]>

.gitignore

@@ -3,7 +3,9 @@ __pycache__/
 .pytest_cache/
 
 # envs
+.venv*/
 env*/
+
 .direnv/
 simulation.env
 .vscode/
@@ -16,6 +18,7 @@ net_viz-*.png
 out.mp4
 websocket_client.log
 logs/
+*.log
 
 #misc
 .coverage

ct-app/README.md

@@ -139,6 +139,7 @@ Parameter | Recommanded value (staging) | Description
 `GCP_FILE_PREFIX` | `expected_reward` | File prefix for GCP distribution list storage
 `GCP_FOLDER` | `staging` | Folder on GCP where to store distribution list
 `PEER_MIN_VERSION` | `2.0.0` | Minimum node version to be eligible
+
 `RABBITMQ_HOST` | (check Bitwarden) | 
 `RABBITMQ_PASSWORD` | (check Bitwarden) | 
 `RABBITMQ_PROJECT_NAME` | `ct-app` | Name of the RabbitMQ project
@@ -179,21 +180,26 @@ Flag | Recommanded value (staging)
 `FLAG_NODE_CLOSE_INCOMING_CHANNELS` (Not available) |--
 `FLAG_NODE_GET_TOTAL_CHANNEL_FUNDS` |--
 
-
 Those flags turn on the corresponding feature if the variable exist. Also, the value associated to the flag defines the delay between two executions of the methods.
 
 #### postman
 This module handles message distribution. It relies on a bunch of parameters:
 
 Parameter | Recommanded value (staging) | Description
 --|--|--
-`PARAM_BATCH_SIZE` | `50` | 
+`PARAM_BATCH_SIZE` | `50` | Number of messages to send before checking the inbox
 `PARAM_DELAY_BETWEEN_TWO_MESSAGES` | `0.25` | Delay between two messages
-`PARAM_MESSAGE_DELIVERY_TIMEOUT` | `10` | Delay between two batches
 `PARAM_MAX_ATTEMPTS` | `4` | Maximum number of retries before timing out
-`RABBITMQ_PROJECT_NAME` | `ct-app` | Name of the RabbitMQ project
+`PARAM_MESSAGE_DELIVERY_TIMEOUT` | `10` | Delay between two batches
+
+#### Common parameters
+In addition to the above-mentioned parameters, there's a bunch of parameters that are required to be able to communicate with databases, RabbitMQ brokers, and nodes.
+Parameter | Recommanded value (staging) | Description
+--|--|--
 `RABBITMQ_HOST` | (check Bitwarden) | 
 `RABBITMQ_PASSWORD` | (check Bitwarden) | 
+`RABBITMQ_PROJECT_NAME` | `ct-app` | Name of the RabbitMQ project
+`RABBITMQ_TASK_NAME` | `fake_task` | Task to create when distributing rewards
 `RABBITMQ_USERNAME` | (check Bitwarden) | 
 `RABBITMQ_VIRTUALHOST` | (check Bitwarden) | 
 `PGHOST` | (from gcloud) |
@@ -205,8 +211,9 @@ Parameter | Recommanded value (staging) | Description
 `PGSSLKEY` |  | Path to the SSL user key
 `PGSSLROOTCERT` |  | Path to the SSL root certificate
 `PGSSLMODE` | `verify-ca` | 
-`NODE_ADDRESS_X` (multiple, min. 2) | (check Bitwarden) |
-`NODE_KEY` | (check Bitwarden) | 
+`NODE_ADDRESS_X` (multiple, min. 2) | (check Bitwarden) | Node endpoints in the format `http://ip:port`
+`NODE_KEY_X` (multiple, min. 2) | (check Bitwarden) | Node API token
+
 
 This program logs to STDOUT. The log level is set to INFO by default.
 

ct-app/core/__main__.py

@@ -35,7 +35,10 @@ def main():
     instance.post_init(nodes, params)
 
     # start the prometheus client
-    start_http_server(8080)
+    try:
+        start_http_server(8080)
+    except OSError:
+        instance.error("Address already in use, prometheus client not started")
 
     loop = asyncio.new_event_loop()
     loop.add_signal_handler(SIGINT, instance.stop)

ct-app/core/components/baseclass.py

@@ -6,6 +6,10 @@
 
 
 class Base:
+    """
+    Base class for logging and printing messages with different colors.
+    """
+
     doLogging = True
 
     handler = logging.StreamHandler()
@@ -31,34 +35,49 @@ def _print(self, message: str, color: str = "\033[0m"):
         print(self.__format(message, color))
 
     def debug(self, message: str):
+        """
+        Log or print a debug message with the specified message.
+        """
         color = "\033[0;32m"
         if self.doLogging:
             self.logger.debug(self.__format(message, color))
         else:
             self._print(message, color)
 
     def info(self, message: str):
+        """
+        Log or print an info message with the specified message.
+        """
         color = "\033[0;34m"
         if self.doLogging:
             self.logger.info(self.__format(message, color))
         else:
             self._print(message, color)
 
     def warning(self, message: str):
+        """
+        Log or print a warning message with the specified message.
+        """
         color = "\033[0;33m"
         if self.doLogging:
             self.logger.warning(self.__format(message, color))
         else:
             self._print(message, color)
 
     def error(self, message: str):
+        """
+        Log or print an error message with the specified message.
+        """
         color = "\033[0;31m"
         if self.doLogging:
             self.logger.error(self.__format(message, color))
         else:
             self._print(message, color)
 
     def feature(self, message: str):
+        """
+        Log or print a feature message with the specified message.
+        """
         color = "\033[0;35m"
         if self.doLogging:
             self.logger.info(self.__format(message, color))

ct-app/core/components/graphql_providers.py

@@ -1,3 +1,4 @@
+import asyncio
 from pathlib import Path
 
 from gql import Client, gql
@@ -54,7 +55,13 @@ async def _test_query(self, key: str, **kwargs) -> bool:
         """
         vars = {"first": 1, "skip": 0}
         vars.update(kwargs)
-        response = await self._execute(self._sku_query, vars)
+
+        # call `self._execute(self._sku_query, vars)` with a timeout
+        try:
+            response = await asyncio.wait_for(self._execute(self._sku_query, vars), timeout=30)
+        except asyncio.TimeoutError:
+            self.error("Query timeout occurred")
+            return False
 
         return response and key in response
 
@@ -73,8 +80,11 @@ async def _get(self, key: str, **kwargs) -> dict:
             vars = {"first": page_size, "skip": skip}
             vars.update(kwargs)
 
-            response = await self._execute(self._sku_query, vars)
-
+            try:
+                response = await asyncio.wait_for(self._execute(self._sku_query, vars), timeout=30)
+            except asyncio.TimeoutError:
+                self.error("Timeout error while fetching data from subgraph.")
+                break            
             if response is None:
                 break