fix: increased median table timeout to 180 seconds + logging for ora sync

Author: cristibleotiu

extensions/business/oracle_sync/sync_mixins/ora_sync_constants.py

@@ -9,6 +9,7 @@
 SIGNATURES_EXCHANGE_MULTIPLIER = 2
 REQUEST_AGREEMENT_TABLE_MULTIPLIER = 5 if DEBUG_MODE else 2
 LOCAL_TABLE_SEND_MULTIPLIER = 3 if DEBUG_MODE else 2
+MEDIAN_TABLE_SEND_MULTIPLIER = 3 if DEBUG_MODE else 2
 
 if DEBUG_MODE:
   SUPERVISOR_MIN_AVAIL_PRC = 0.4

extensions/business/oracle_sync/sync_mixins/ora_sync_states_mixin.py

@@ -9,6 +9,7 @@
   ORACLE_SYNC_ACCEPTED_MEDIAN_ERROR_MARGIN,
 
   LOCAL_TABLE_SEND_MULTIPLIER,
+  MEDIAN_TABLE_SEND_MULTIPLIER,
   REQUEST_AGREEMENT_TABLE_MULTIPLIER,
   SIGNATURES_EXCHANGE_MULTIPLIER,
 )
@@ -509,6 +510,9 @@ def _receive_local_table_and_maybe_send_local_table(self):
       # end for
       return
 
+    def get_local_table_timeout(self):
+      return self.cfg_send_period * LOCAL_TABLE_SEND_MULTIPLIER
+
     def _send_local_table_timeout(self):
       """
       Check if the exchange phase of the local table has finished.
@@ -517,8 +521,8 @@ def _send_local_table_timeout(self):
       -------
       bool: True if the exchange phase of the local table has finished, False otherwise
       """
-      timeout_reached = (self.time() - self.first_time_local_table_sent) > (
-            self.cfg_send_period * LOCAL_TABLE_SEND_MULTIPLIER)
+      first_time = self.first_time_local_table_sent
+      timeout_reached = first_time is not None and (self.time() - first_time) > self.get_local_table_timeout()
       early_stopping = self._maybe_early_stop_phase(
         data=self.dct_local_tables,
         phase=self.STATES.S2_SEND_LOCAL_TABLE,
@@ -713,6 +717,9 @@ def _receive_median_table_and_maybe_send_median_table(self):
       # end for
       return
 
+    def get_median_table_timeout(self):
+      return self.cfg_send_period * MEDIAN_TABLE_SEND_MULTIPLIER
+
     def _send_median_table_timeout(self):
       """
       Check if the exchange phase of the median table has finished.
@@ -721,7 +728,8 @@ def _send_median_table_timeout(self):
       -------
       bool: True if the exchange phase of the median table has finished, False otherwise
       """
-      timeout_reached = (self.time() - self.first_time_median_table_sent) > self.cfg_send_period
+      first_time = self.first_time_median_table_sent
+      timeout_reached = first_time is not None and (self.time() - first_time) > self.get_median_table_timeout()
       early_stopping = self._maybe_early_stop_phase(
         data=self.dct_median_tables,
         phase=self.STATES.S4_SEND_MEDIAN_TABLE,
@@ -1020,6 +1028,9 @@ def _receive_agreement_signature_and_maybe_send_agreement_signature(self):
       # endfor received messages
       return
 
+    def get_agreement_signature_timeout(self):
+      return self.cfg_send_period
+
     def _send_agreement_signature_timeout(self):
       """
       Check if the exchange phase of the agreed median table has finished.
@@ -1028,7 +1039,8 @@ def _send_agreement_signature_timeout(self):
       -------
       bool: True if the exchange phase of the agreed median table has finished, False otherwise
       """
-      timeout_reached = (self.time() - self.first_time__agreement_signature_sent) > self.cfg_send_period
+      first_time = self.first_time__agreement_signature_sent
+      timeout_reached = first_time is not None and (self.time() - first_time) > self.get_agreement_signature_timeout()
       early_stopping = self._maybe_early_stop_phase(
         data=self.compiled_agreed_median_table_signatures,
         phase=self.STATES.S6_SEND_AGREED_MEDIAN_TABLE,
@@ -1086,6 +1098,9 @@ def _exchange_agreement_signatures(self):
       # endfor received messages
       return
 
+    def get_exchange_signatures_timeout(self):
+      return self.cfg_send_period * SIGNATURES_EXCHANGE_MULTIPLIER
+
     def _exchange_signatures_timeout(self):
       """
       Check if the exchange phase of the agreement signatures has finished.
@@ -1094,8 +1109,8 @@ def _exchange_signatures_timeout(self):
       -------
       bool: True if the exchange phase of the agreement signatures has finished, False otherwise
       """
-      timeout_reached = (self.time() - self.first_time__agreement_signatures_exchanged) > (
-          self.cfg_send_period * SIGNATURES_EXCHANGE_MULTIPLIER)
+      first_time = self.first_time__agreement_signatures_exchanged
+      timeout_reached = first_time is not None and (self.time() - first_time) > self.get_exchange_signatures_timeout()
       early_stopping = self._maybe_early_stop_phase(
         data=self.compiled_agreed_median_table_signatures,
         phase=self.STATES.S10_EXCHANGE_AGREEMENT_SIGNATURES,

extensions/business/oracle_sync/sync_mixins/ora_sync_utils_mixin.py

@@ -332,9 +332,14 @@ def update_participant_oracles(
       # endfor
       total_current_participants = sum(self.is_participating.values())
       if state is not None:
+        elapsed, total = self.get_elapsed_and_total_time_of_stage(stage=state)
+        elapsed_str = ""
+        if elapsed_str is not None and total is not None:
+          elapsed_str = f"[elapsed: {elapsed:.2f}s/{total:.2f}s]"
+        # endif elapsed and total
         log_str = f"Updating participating oracles based on {state = }.\n"
         log_str += f"{total_previous_participants} previous participants => "
-        log_str += f"{total_current_participants} current participants.\n"
+        log_str += f"{total_current_participants} current participants.{elapsed_str}\n"
         if len(disappearing_oracles) > 0:
           log_str += f"{len(disappearing_oracles)} oracles disappeared:"
           log_str += "".join([
@@ -568,7 +573,52 @@ def get_sender_str(self, sender: str):
       -------
       str : The string representation of the sender
       """
-      return f"`{self.netmon.network_node_eeid(sender)}` <{sender}>"
+      short_sender = self.shorten_address(sender)
+      return f"`{self.netmon.network_node_eeid(sender)}` <{short_sender}>"
+
+    def get_elapsed_and_total_time_of_stage(self, stage: str):
+      """
+      Get the elapsed and total time of a stage.
+      Parameters
+      ----------
+      stage : str
+          The stage to get the elapsed and total time for.
+
+      Returns
+      -------
+      (elapsed, total) : tuple
+          The elapsed and total time of the stage.
+          If the stage is not found or the stage does not have a timeout period, (None, None) is returned.
+      """
+      elapsed = None
+      total = None
+      values = {
+        self.STATES.S2_SEND_LOCAL_TABLE: {
+          'started': self.first_time_local_table_sent,
+          'total': self.get_local_table_timeout()
+        },
+        self.STATES.S4_SEND_MEDIAN_TABLE: {
+          'started': self.first_time_median_table_sent,
+          'total': self.get_median_table_timeout()
+        },
+        self.STATES.S6_SEND_AGREED_MEDIAN_TABLE: {
+          'started': self.first_time__agreement_signature_sent,
+          'total': self.get_agreement_signature_timeout()
+        },
+        self.STATES.S10_EXCHANGE_AGREEMENT_SIGNATURES: {
+          'started': self.first_time__agreement_signatures_exchanged,
+          'total': self.get_exchange_signatures_timeout()
+        }
+      }
+      stage_values = values.get(stage)
+      if stage_values is not None:
+        started = stage_values.get('started')
+        total = stage_values.get('total')
+        if started is not None:
+          elapsed = self.time() - started
+        # endif started exists
+      # endif stage_values is not None
+      return elapsed, total
 
     def log_received_message(
         self,
@@ -584,6 +634,12 @@ def log_received_message(
       sender_str = self.get_sender_str(sender)
       log_str = f"{progress_str}Received message{duplicated_str} from oracle {sender_str}: {stage = }"
 
+      elapsed, total = self.get_elapsed_and_total_time_of_stage(stage)
+
+      if elapsed is not None and total is not None:
+        log_str += f" | elapsed: {elapsed:.2f}s/{total:.2f}s"
+      # endif elapsed and total
+
       if return_str:
         return log_str
       self.P(log_str)

ver.py

@@ -1,2 +1,2 @@
-__VER__ = '2.9.680'
+__VER__ = '2.9.690'