Better error observability (#1172)

* do not give metrics error if fetching stopped tasks fails

* init better logging

* do not return error from optional message fetcher

* fmt

* change log level

* check number sent messages in aggregator rpc streams, log errors

* change comment

* catch more errors

* fix wrong num needed sigs issue

* better errors on nonce aggregation

* add partial signature verification

* verifty partial sigs only when env var is true

* change var naming

* update docs

* add logging

* show all partial verify errors, not only the first

* fix tracing errors

* fix typos

* refactor, better err formatting

* show verifier pks instead of index (because its ordered) in several error messages

* Update core/src/rpc/aggregator.rs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* better error logging in aggregator setup()

* adddress copilot comments

* Update core/src/rpc/aggregator.rs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* ignore test

* fix grammar

* monitor operator deposit_sign

* add observing errors of every verifier in opt_payout_sign, add timeout to op_payout_sign

* Update core/src/rpc/verifier.rs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* cargo fmt

* rename var

* address comments

* rename fn

* refactor flatten_join_named_results

* cargo fmt

* adjust test

* update test again

* add serial to test

* increase timeout

* skip a compatibility test in debug mode

---------

Co-authored-by: atacann <atacaniyidogan@gmail.com>
Co-authored-by: Mehmet Efe Akça <efeakca@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: atacann <111396231+atacann@users.noreply.github.com>

Author: 4 people

core/src/aggregator.rs

@@ -429,7 +429,9 @@ impl Aggregator {
                                 debug_span!("get_deposit_keys", id=%OperatorId(operator_xonly_pk)),
                             )
                             .await
-                            .wrap_err(Status::internal("Operator key retrieval failed"))?
+                            .wrap_err(Status::internal(format!(
+                                "Operator {operator_xonly_pk} key retrieval failed"
+                            )))?
                             .into_inner();
 
                         // A send error means that all receivers are closed,

core/src/compatibility.rs

@@ -276,8 +276,15 @@ mod tests {
         assert!(msg.contains("Security council mismatch"));
     }
 
-    #[tokio::test]
+    // serial test because it calculates sha256 of the bitvm cache for all actors
+    #[tokio::test(flavor = "multi_thread")]
     async fn test_get_compatibility_data_from_entities() {
+        // Skip this test if running in debug mode, hashing bitvm cache is slow
+        if cfg!(debug_assertions) {
+            eprintln!("Skipping test_get_compatibility_data_from_entities in debug mode");
+            return;
+        }
+
         let mut config = create_test_config_with_thread_name().await;
         let _regtest = create_regtest_rpc(&mut config).await;
         let actors = create_actors::<MockCitreaClient>(&config).await;

core/src/constants.rs

@@ -65,16 +65,20 @@ mod timeout {
     pub const OPERATOR_SIGS_STREAM_CREATION_TIMEOUT: Duration = Duration::from_secs(300); // 5 minutes
     pub const DEPOSIT_FINALIZE_STREAM_CREATION_TIMEOUT: Duration = Duration::from_secs(300); // 5 minutes
 
+    pub const SETUP_COMPLETION_TIMEOUT: Duration = Duration::from_secs(1800); // 30 minutes
+
     pub const PIPELINE_COMPLETION_TIMEOUT: Duration = Duration::from_secs(3600); // 60 minutes
     pub const OPERATOR_SIGS_TIMEOUT: Duration = Duration::from_secs(1200); // 20 minutes
     pub const SEND_OPERATOR_SIGS_TIMEOUT: Duration = Duration::from_secs(600); // 10 minutes
     pub const DEPOSIT_FINALIZATION_TIMEOUT: Duration = Duration::from_secs(2400); // 40 minutes
 
+    pub const OPTIMISTIC_PAYOUT_TIMEOUT: Duration = Duration::from_secs(300); // 5 minutes
+
     pub const RESTART_BACKGROUND_TASKS_TIMEOUT: Duration = Duration::from_secs(120);
 
     pub const ENTITY_STATUS_POLL_TIMEOUT: Duration = Duration::from_secs(120);
 
-    pub const ENTITY_COMP_DATA_POLL_TIMEOUT: Duration = Duration::from_secs(120); // 2 minutes
+    pub const ENTITY_COMP_DATA_POLL_TIMEOUT: Duration = Duration::from_secs(240); // 4 minutes
 
     pub const PUBLIC_KEY_COLLECTION_TIMEOUT: Duration = Duration::from_secs(30);
 

core/src/errors.rs

@@ -252,27 +252,22 @@ impl From<Status> for BridgeError {
 
 impl From<BridgeError> for tonic::Status {
     fn from(val: BridgeError) -> Self {
-        let eyre_report = val.into_eyre();
-
-        // eyre::Report can cast any error in the chain to a Status, so we use its downcast method to get the first Status.
-        eyre_report.downcast::<Status>().unwrap_or_else(|report| {
-            // We don't want this case to happen, all casts to Status should contain a Status that contains a user-facing error message.
-            tracing::error!(
-                "Returning internal error on RPC call, full error: {:?}",
-                report
-            );
-
-            let mut status = tonic::Status::internal(report.to_string());
-            status.set_source(Into::into(
-                Into::<Box<dyn std::error::Error + Send + Sync>>::into(report),
-            ));
-            status
-        })
+        let err = format!("{val:#}");
+        // delete escape characters
+        let flattened = err
+            .replace("\\n", " ") // remove escaped newlines
+            .replace("\n", " ") // remove real newlines
+            .replace("\"", "") // delete quotes
+            .replace("\\", ""); // remove any remaining backslashes
+        let whitespace_removed = flattened.split_whitespace().collect::<Vec<_>>().join(" ");
+        tonic::Status::internal(whitespace_removed)
     }
 }
 
 #[cfg(test)]
 mod tests {
+    use eyre::Context;
+
     use super::*;
     #[test]
     fn test_downcast() {
@@ -290,14 +285,18 @@ mod tests {
     }
 
     #[test]
-    fn test_status_in_chain_cast_properly() {
-        let err: BridgeError = eyre::eyre!("Some problem")
-            .wrap_err(tonic::Status::deadline_exceeded("Some timer expired"))
-            .wrap_err("Something else went wrong")
-            .into();
+    fn test_status_shows_all_errors_in_chain() {
+        let err: BridgeError = Err::<(), BridgeError>(BridgeError::BitcoinRPC(
+            BitcoinRPCError::TransactionNotConfirmed,
+        ))
+        .wrap_err(tonic::Status::deadline_exceeded("Error A"))
+        .wrap_err("Error B")
+        .unwrap_err()
+        .into();
 
-        let status: Status = err.into_status();
-        assert_eq!(status.code(), tonic::Code::DeadlineExceeded);
-        assert_eq!(status.message(), "Some timer expired");
+        let status: Status = err.into();
+        assert!(status.message().contains("Error A"));
+        assert!(status.message().contains("Error B"));
+        assert!(status.message().contains("Bitcoin"));
     }
 }

core/src/musig2.rs

@@ -2,7 +2,7 @@
 //!
 //! Helper functions for the MuSig2 signature scheme.
 
-use crate::{bitvm_client::SECP, errors::BridgeError};
+use crate::{aggregator::VerifierId, bitvm_client::SECP, errors::BridgeError};
 use bitcoin::{
     hashes::Hash,
     key::Keypair,
@@ -195,6 +195,7 @@ pub fn aggregate_partial_signatures(
         .unwrap_or(false);
 
     if enable_partial_sig_verification {
+        let mut partial_sig_verification_errors = Vec::new();
         for ((partial_sig, pub_nonce), pub_key) in
             partial_sigs_and_nonces.into_iter().zip(pks.into_iter())
         {
@@ -205,16 +206,18 @@ pub fn aggregate_partial_signatures(
                 *pub_nonce,
                 to_secp_pk(pub_key),
             ) {
-                tracing::error!(
-                    "MuSig2 Error: partial signature verification failed for pub key: {}",
-                    pub_key
-                );
-                return Err(BridgeError::from(eyre::eyre!(
-                    "MuSig2 Error: partial signature verification failed for pub key: {}",
-                    pub_key
-                )));
+                let verifier_id = VerifierId(pub_key);
+                partial_sig_verification_errors.push(format!("{verifier_id}"));
             }
         }
+        if !partial_sig_verification_errors.is_empty() {
+            let error_msg = format!(
+                "MuSig2 Error: partial signature verification failed for verifiers: {}",
+                partial_sig_verification_errors.join(", ")
+            );
+            tracing::error!("{error_msg}");
+            return Err(BridgeError::from(eyre::eyre!(error_msg)));
+        }
     }
     let final_sig = session.partial_sig_agg(&partial_sigs);
 

core/src/operator.rs

@@ -20,7 +20,7 @@ use crate::errors::BridgeError;
 use crate::extended_bitcoin_rpc::ExtendedBitcoinRpc;
 
 use crate::metrics::L1SyncStatusProvider;
-use crate::rpc::clementine::EntityStatus;
+use crate::rpc::clementine::{EntityStatus, StoppedTasks};
 use crate::task::entity_metric_publisher::{
     EntityMetricPublisher, ENTITY_METRIC_PUBLISHER_INTERVAL,
 };
@@ -225,7 +225,16 @@ where
     }
 
     pub async fn get_current_status(&self) -> Result<EntityStatus, BridgeError> {
-        let stopped_tasks = self.background_tasks.get_stopped_tasks().await?;
+        let stopped_tasks = match self.background_tasks.get_stopped_tasks().await {
+            Ok(stopped_tasks) => stopped_tasks,
+            Err(e) => {
+                tracing::error!("Failed to get stopped tasks: {:?}", e);
+                StoppedTasks {
+                    stopped_tasks: vec![format!("Stopped tasks fetch failed {:?}", e)],
+                }
+            }
+        };
+
         // Determine if automation is enabled
         let automation_enabled = cfg!(feature = "automation");
 
@@ -479,13 +488,14 @@ where
     pub async fn deposit_sign(
         &self,
         mut deposit_data: DepositData,
-    ) -> Result<mpsc::Receiver<schnorr::Signature>, BridgeError> {
+    ) -> Result<mpsc::Receiver<Result<schnorr::Signature, BridgeError>>, BridgeError> {
         self.citrea_client
             .check_nofn_correctness(deposit_data.get_nofn_xonly_pk()?)
             .await?;
 
         let mut tweak_cache = TweakCache::default();
         let (sig_tx, sig_rx) = mpsc::channel(constants::DEFAULT_CHANNEL_SIZE);
+        let monitor_err_sender = sig_tx.clone();
 
         let deposit_blockhash = self
             .rpc
@@ -511,14 +521,15 @@ where
                     Some(&mut tweak_cache),
                 )?;
 
-                if sig_tx.send(sig).await.is_err() {
-                    break;
-                }
+                sig_tx
+                    .send(Ok(sig))
+                    .await
+                    .wrap_err("Failed to send signature in operator deposit sign")?;
             }
 
             Ok::<(), BridgeError>(())
         });
-        monitor_standalone_task(handle, "Operator deposit sign");
+        monitor_standalone_task(handle, "Operator deposit sign", monitor_err_sender);
 
         Ok(sig_rx)
     }