retry when artifact creation failed

Author: ghubertpalo

mithril-aggregator/src/runtime/error.rs

@@ -1,4 +1,4 @@
-use std::error::Error as StdError;
+use mithril_common::StdError;
 use thiserror::Error;
 
 /// Error encountered or produced by the Runtime.
@@ -13,7 +13,7 @@ pub enum RuntimeError {
         message: String,
 
         /// Eventual caught error
-        nested_error: Option<Box<dyn StdError + Sync + Send>>,
+        nested_error: Option<StdError>,
     },
     /// A Critical error means the Runtime stops and the software exits with an
     /// error code.
@@ -23,30 +23,39 @@ pub enum RuntimeError {
         message: String,
 
         /// Eventual caught error
-        nested_error: Option<Box<dyn StdError + Sync + Send>>,
+        nested_error: Option<StdError>,
+    },
+    /// An error that needs to re-initialize the state machine.
+    #[error("An error occured: {message}. The state machine will be re-initialized. Nested error: {nested_error:#?}")]
+    ReInit {
+        /// error message
+        message: String,
+
+        /// Eventual caught error
+        nested_error: Option<StdError>,
     },
 }
 
 impl RuntimeError {
     /// Create a new KeepState error
-    pub fn keep_state(message: &str, error: Option<Box<dyn StdError + Sync + Send>>) -> Self {
+    pub fn keep_state(message: &str, error: Option<StdError>) -> Self {
         Self::KeepState {
             message: message.to_string(),
             nested_error: error,
         }
     }
 
     /// Create a new Critical error
-    pub fn critical(message: &str, error: Option<Box<dyn StdError + Sync + Send>>) -> Self {
+    pub fn critical(message: &str, error: Option<StdError>) -> Self {
         Self::Critical {
             message: message.to_string(),
             nested_error: error,
         }
     }
 }
 
-impl From<Box<dyn StdError + Sync + Send>> for RuntimeError {
-    fn from(value: Box<dyn StdError + Sync + Send>) -> Self {
+impl From<StdError> for RuntimeError {
+    fn from(value: StdError) -> Self {
         Self::KeepState {
             message: "Error caught, state preserved, will retry to cycle.".to_string(),
             nested_error: Some(value),

mithril-aggregator/src/runtime/runner.rs

@@ -197,26 +197,32 @@ impl AggregatorRunnerTrait for AggregatorRunner {
                     "RUNNER: get_current_non_certified_open_message_for_signed_entity_type: existing open message found, not already certified";
                     "signed_entity_type" => ?signed_entity_type
                 );
+
                 Some(existing_open_message)
             }
             Some(_) => {
                 info!(
                     "RUNNER: get_current_non_certified_open_message_for_signed_entity_type: existing open message found, already certified";
                     "signed_entity_type" => ?signed_entity_type
                 );
+
                 None
             }
             None => {
                 info!(
                     "RUNNER: get_current_non_certified_open_message_for_signed_entity_type: no open message found, a new one will be created";
                     "signed_entity_type" => ?signed_entity_type
                 );
-                // In order to fix flakiness in the end to end tests, we are compelled to update the beacon (of the multi-signer)
-                // This avoids the computation of the next AVK on the wrong epoch (in 'compute_protocol_message')
-                // TODO: remove 'current_beacon' from the multi-signer state which will avoid this fix
+                // In order to fix flakiness in the end to end tests, we are
+                // compelled to update the beacon (of the multi-signer) This
+                // avoids the computation of the next AVK on the wrong epoch (in
+                // 'compute_protocol_message')
+                // TODO: remove 'current_beacon' from the multi-signer state
+                // which will avoid this fix
                 let chain_beacon: Beacon = self.get_beacon_from_chain().await?;
                 self.update_beacon(&chain_beacon).await?;
                 let protocol_message = self.compute_protocol_message(signed_entity_type).await?;
+
                 Some(
                     self.create_open_message(signed_entity_type, &protocol_message)
                         .await?,

mithril-aggregator/src/runtime/state_machine.rs

@@ -51,7 +51,9 @@ impl Display for AggregatorState {
 
 /// The AggregatorRuntime responsibility is to create a state machine to handle
 /// all actions required by the process of getting multi-signatures.
-/// See the [documentation](https://mithril.network/doc/mithril/mithril-network/aggregator#under-the-hood) for more explanations about the Aggregator state machine.
+/// See the
+/// [documentation](https://mithril.network/doc/mithril/mithril-network/aggregator#under-the-hood)
+/// for more explanations about the Aggregator state machine.
 pub struct AggregatorRuntime {
     /// the internal state of the automate
     state: AggregatorState,
@@ -104,6 +106,8 @@ impl AggregatorRuntime {
 
         loop {
             if let Err(e) = self.cycle().await {
+                warn!("State machine issued an error: {e}");
+
                 match &e {
                     RuntimeError::Critical {
                         message: _,
@@ -113,8 +117,32 @@ impl AggregatorRuntime {
 
                         return Err(e);
                     }
-                    _ => {
-                        warn!("state machine: cycle returned an error: '{e}'.");
+                    RuntimeError::KeepState {
+                        message,
+                        nested_error,
+                    } => {
+                        warn!(
+                            "KeepState Error: {message}. Nested error: «{}».",
+                            nested_error
+                                .as_ref()
+                                .map(|e| format!("{e}"))
+                                .unwrap_or("None".into())
+                        );
+                    }
+                    RuntimeError::ReInit {
+                        message,
+                        nested_error,
+                    } => {
+                        warn!(
+                            "ReInit Error: {message}. Nested error: «{}».",
+                            nested_error
+                                .as_ref()
+                                .map(|e| format!("{e}"))
+                                .unwrap_or("None".into())
+                        );
+                        self.state = AggregatorState::Idle(IdleState {
+                            current_beacon: None,
+                        });
                     }
                 }
             }
@@ -280,23 +308,21 @@ impl AggregatorRuntime {
                 message: "not enough signature yet to create a certificate, waiting…".to_string(),
                 nested_error: None,
             })?;
-
-        // Every error raised below this comment must be raised to critical in order to trigger an aggregator restart which will avoid a dead loop in the state machine
-        // TODO: Implement a graceful retry mechanism for artifact creation and label open messages with `is_artifact_created` (to keep track of certified but without artifacts open messages)
-        self.runner.drop_pending_certificate().await.map_err(|e| {
-            RuntimeError::critical(
-                "transiting SIGNING → READY: failed dropping pending certificate",
-                Some(e),
-            )
-        })?;
+        self.runner
+            .drop_pending_certificate()
+            .await
+            .map_err(|e| RuntimeError::ReInit {
+                message: "transiting SIGNING → READY: failed to drop pending certificate"
+                    .to_string(),
+                nested_error: Some(e),
+            })?;
         self.runner
             .create_artifact(&state.open_message.signed_entity_type, &certificate)
             .await
-            .map_err(|e| {
-                RuntimeError::critical(
-                    "transiting SIGNING → READY: failed creating artifact",
-                    Some(e),
-                )
+            .map_err(|e| RuntimeError::ReInit {
+                message: "transiting SIGNING → READY: failed to create artifact. Retrying…"
+                    .to_string(),
+                nested_error: Some(e),
             })?;
 
         Ok(ReadyState {
@@ -653,11 +679,62 @@ mod tests {
             open_message: OpenMessage::dummy(),
         };
         let mut runtime = init_runtime(Some(AggregatorState::Signing(state)), runner).await;
-        runtime
+        let err = runtime
             .cycle()
             .await
             .expect_err("cycle should have returned an error");
 
+        match err {
+            RuntimeError::KeepState {
+                message: _,
+                nested_error: _,
+            } => (),
+            _ => panic!("KeepState error expected, got {err:?}."),
+        };
+
+        assert_eq!("signing".to_string(), runtime.get_state());
+    }
+
+    #[tokio::test]
+    async fn signing_artifact_not_created() {
+        let mut runner = MockAggregatorRunner::new();
+        runner
+            .expect_get_beacon_from_chain()
+            .once()
+            .returning(|| Ok(fake_data::beacon()));
+        runner
+            .expect_get_current_non_certified_open_message_for_signed_entity_type()
+            .once()
+            .returning(|_| Ok(Some(OpenMessage::dummy())));
+        runner
+            .expect_create_certificate()
+            .return_once(move |_| Ok(Some(fake_data::certificate("whatever".to_string()))));
+        runner
+            .expect_drop_pending_certificate()
+            .once()
+            .returning(|| Ok(Some(fake_data::certificate_pending())));
+        runner
+            .expect_create_artifact()
+            .once()
+            .returning(|_, _| Err("whatever".into()));
+        let state = SigningState {
+            current_beacon: fake_data::beacon(),
+            open_message: OpenMessage::dummy(),
+        };
+        let mut runtime = init_runtime(Some(AggregatorState::Signing(state)), runner).await;
+        let err = runtime
+            .cycle()
+            .await
+            .expect_err("cycle should have returned an error");
+
+        match err {
+            RuntimeError::ReInit {
+                message: _,
+                nested_error: _,
+            } => (),
+            _ => panic!("ReInit error expected, got {err:?}."),
+        };
+
         assert_eq!("signing".to_string(), runtime.get_state());
     }
 

mithril-aggregator/src/signed_entity_service.rs

@@ -120,9 +120,20 @@ impl SignedEntityService for MithrilSignedEntityService {
             "certificate_hash" => &certificate.hash
         );
 
-        let artifact = self
-            .compute_artifact(signed_entity_type.clone(), certificate)
-            .await?;
+        let mut remaining_retries = 3;
+        let artifact = loop {
+            remaining_retries -= 1;
+
+            match self
+                .compute_artifact(signed_entity_type.clone(), certificate)
+                .await
+            {
+                Err(error) if remaining_retries == 0 => break Err(error),
+                Err(_error) => (),
+                Ok(artifact) => break Ok(artifact),
+            };
+        }?;
+
         let signed_entity = SignedEntityRecord {
             signed_entity_id: artifact.get_id(),
             signed_entity_type,