feat(aggregator): add delay on error in signature processor

Author: jpraynaud

mithril-aggregator/src/services/signature_processor.rs

@@ -1,5 +1,6 @@
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
+use anyhow::{Context, anyhow};
 use slog::{Logger, error, trace, warn};
 
 use mithril_common::{
@@ -34,6 +35,8 @@ pub struct SequentialSignatureProcessor {
 }
 
 impl SequentialSignatureProcessor {
+    const ERROR_DELAY_IN_SECONDS: Duration = Duration::from_secs(10);
+
     /// Creates a new `SignatureProcessor` instance.
     pub fn new(
         consumer: Arc<dyn SignatureConsumer>,
@@ -64,6 +67,7 @@ impl SignatureProcessor for SequentialSignatureProcessor {
     async fn process_signatures(&self) -> StdResult<()> {
         let origin_network = self.consumer.get_origin_tag();
 
+        let mut total_import_errors = 0;
         match self.consumer.get_signatures().await {
             Ok(signatures) => {
                 let number_of_signatures = signatures.len() as u32;
@@ -86,6 +90,7 @@ impl SignatureProcessor for SequentialSignatureProcessor {
                                 .increment(&[&origin_network]);
                         }
                         Err(e) => {
+                            total_import_errors += 1;
                             error!(
                                 self.logger, "Error dispatching single signature";
                                 "full_payload" => #?signature, "error" => ?e
@@ -96,9 +101,24 @@ impl SignatureProcessor for SequentialSignatureProcessor {
             }
             Err(e) => {
                 error!(self.logger, "Error consuming single signatures"; "error" => ?e);
+                total_import_errors += 1;
             }
         }
 
+        if total_import_errors > 0 {
+            error!(
+                self.logger,
+                "Total import errors while processing signatures: {}", total_import_errors
+            );
+            return Err(anyhow!(
+                "Total import errors while processing signatures: {}",
+                total_import_errors
+            ))
+            .with_context(
+                || "SequentialSignatureProcessor encountered errors while processing signatures",
+            );
+        }
+
         Ok(())
     }
 
@@ -111,7 +131,13 @@ impl SignatureProcessor for SequentialSignatureProcessor {
 
                     return Ok(());
                 }
-                _ = self.process_signatures() => {}
+                res = self.process_signatures() => {
+                    if let Err(e) = res {
+                        error!(self.logger, "Error processing signatures"; "error" => ?e);
+                        error!(self.logger, "Sleep for {} seconds", Self::ERROR_DELAY_IN_SECONDS.as_secs());
+                        tokio::time::sleep(Self::ERROR_DELAY_IN_SECONDS).await;
+                    }
+                }
             }
         }
     }