chore: refactor/fix logs for orchestrator (#858)

Co-authored-by: Heemank Verma <[email protected]>
Co-authored-by: Mohit Dhattarwal <[email protected]>

Author: 3 people

madara/crates/client/gateway/client/src/methods.rs

@@ -65,7 +65,8 @@ impl GatewayProvider {
                 .with_block_id(&block_id);
 
             request.send_get::<ProviderBlock>().await
-        }).await
+        })
+        .await
     }
 
     pub async fn get_preconfirmed_block(&self, block_number: u64) -> Result<ProviderBlockPreConfirmed, SequencerError> {
@@ -76,7 +77,8 @@ impl GatewayProvider {
                 .with_block_id(&BlockId::Number(block_number));
 
             request.send_get::<ProviderBlockPreConfirmed>().await
-        }).await
+        })
+        .await
     }
 
     pub async fn get_header(&self, block_id: BlockId) -> Result<ProviderBlockHeader, SequencerError> {
@@ -88,7 +90,8 @@ impl GatewayProvider {
                 .add_param("headerOnly", "true");
 
             request.send_get::<ProviderBlockHeader>().await
-        }).await
+        })
+        .await
     }
 
     pub async fn get_state_update(&self, block_id: BlockId) -> Result<ProviderStateUpdate, SequencerError> {
@@ -99,7 +102,8 @@ impl GatewayProvider {
                 .with_block_id(&block_id);
 
             request.send_get::<ProviderStateUpdate>().await
-        }).await
+        })
+        .await
     }
 
     pub async fn get_block_bouncer_weights(&self, block_number: u64) -> Result<BouncerWeights, SequencerError> {
@@ -123,7 +127,8 @@ impl GatewayProvider {
                 .add_param(Cow::from("includeBlock"), "true");
 
             request.send_get::<ProviderStateUpdateWithBlock>().await
-        }).await
+        })
+        .await
     }
 
     pub async fn get_signature(&self, block_id: BlockId) -> Result<ProviderBlockSignature, SequencerError> {
@@ -138,7 +143,8 @@ impl GatewayProvider {
                 .with_block_id(&block_id);
 
             request.send_get::<ProviderBlockSignature>().await
-        }).await
+        })
+        .await
     }
 
     pub async fn get_class_by_hash(
@@ -165,7 +171,8 @@ impl GatewayProvider {
                 let err = serde::de::Error::custom("Unknown contract type".to_string());
                 Err(SequencerError::DeserializeBody { serde_error: err })
             }
-        }).await
+        })
+        .await
     }
 
     async fn add_transaction<T>(&self, transaction: UserTransaction) -> Result<T, SequencerError>
@@ -421,6 +428,7 @@ mod tests {
 
     #[rstest]
     #[tokio::test]
+    #[ignore = "Madara is incompatible with 0.14.1 feeder gateway changes, specifically migrated_compiled_classes."]
     async fn get_preconfirmed_block(client_testnet_fixture: GatewayProvider) {
         let latest_block_number =
             client_testnet_fixture.get_header(BlockId::Tag(BlockTag::Latest)).await.unwrap().block_number;

orchestrator/crates/settlement-clients/ethereum/src/lib.rs

@@ -43,7 +43,7 @@ use lazy_static::lazy_static;
 use mockall::automock;
 use reqwest::Client;
 use tokio::time::sleep;
-use tracing::{info, warn};
+use tracing::{error, info, warn};
 
 // For more details on state update, refer to the core contract logic
 // https://github.com/starkware-libs/cairo-lang/blob/master/src/starkware/starknet/solidity/Output.sol
@@ -263,11 +263,13 @@ impl SettlementClient for EthereumSettlementClient {
         state_diff: Vec<Vec<u8>>,
         _nonce: u64,
     ) -> Result<String> {
-        tracing::info!(
+        // TODO(prakhar,20/11/2025): Update the logs to add custom formatter - https://github.com/madara-alliance/madara/blob/d2a1e8050a3d01ccf398f57616cbc4fb6386aaa6/madara/crates/client/analytics/src/formatter.rs#L288
+        info!(
             log_type = "starting",
             category = "update_state",
-            function_type = "blobs",
-            "Updating state with blobs."
+            state_diff_len = %state_diff.len(),
+            program_output_len = %program_output.len(),
+            "Updating state with blob"
         );
 
         let mut mul_factor = GAS_PRICE_MULTIPLIER_START;
@@ -279,8 +281,15 @@ impl SettlementClient for EthereumSettlementClient {
                 Result::Ok(pending_transaction) => pending_transaction,
                 Err(e) => match e {
                     SendTransactionError::ReplacementTransactionUnderpriced(e) => {
-                        warn!("Failed to send the state update transaction: {:?} with {:?} multiplication factor, trying again...", e, mul_factor);
-                        mul_factor = self.get_next_mul_factor(mul_factor)?;
+                        let next_mul_factor = self.get_next_mul_factor(mul_factor)?;
+                        warn!(
+                            current_multiplier = %mul_factor,
+                            next_multiplier = %next_mul_factor,
+                            max_multiplier = %self.max_gas_price_mul_factor,
+                            error = ?e,
+                            "Transaction rejected due to low gas price, retrying with higher multiplier"
+                        );
+                        mul_factor = next_mul_factor;
                         continue;
                     }
                     SendTransactionError::Other(_) => {
@@ -289,24 +298,29 @@ impl SettlementClient for EthereumSettlementClient {
                 },
             };
 
-            tracing::info!(
+            info!(
                 log_type = "completed",
                 category = "update_state",
                 function_type = "blobs",
-                "State updated with blobs."
+                tx_hash = %pending_transaction.tx_hash(),
+                "State update transaction submitted to Ethereum with blobs"
             );
 
-            tracing::warn!("⏳ Waiting for txn finality...");
-
             // Waiting for transaction finality
             let res = self.wait_for_tx_finality(&pending_transaction.tx_hash().to_string()).await?;
 
             match res {
                 Some(_) => {
-                    tracing::info!("✅ Txn hash : {:?} finalized", pending_transaction.tx_hash().to_string());
+                    info!(
+                        tx_hash = %pending_transaction.tx_hash(),
+                        "Transaction finalized successfully"
+                    );
                 }
                 None => {
-                    tracing::error!("❌ Txn hash: {:?} not finalised", pending_transaction.tx_hash().to_string());
+                    error!(
+                        tx_hash = %pending_transaction.tx_hash(),
+                        "Transaction not finalized"
+                    );
                 }
             }
             return Ok(pending_transaction.tx_hash().to_string());

orchestrator/crates/settlement-clients/starknet/src/lib.rs

@@ -24,6 +24,7 @@ use starknet::providers::jsonrpc::HttpTransport;
 use starknet::providers::{JsonRpcClient, Provider};
 use starknet::signers::{LocalWallet, SigningKey};
 use tokio::time::{sleep, Duration};
+use tracing::info;
 
 use crate::conversion::{slice_slice_u8_to_vec_field, u64_from_felt};
 use crate::utils::LocalWalletSignerMiddleware;
@@ -122,11 +123,12 @@ impl SettlementClient for StarknetSettlementClient {
         _onchain_data_hash: [u8; 32],
         _onchain_data_size: [u8; 32],
     ) -> Result<String> {
-        tracing::info!(
+        info!(
             log_type = "starting",
             category = "update_state",
-            function_type = "calldata",
-            "Updating state with calldata."
+            snos_output_len = %snos_output.len(),
+            program_output_len = %program_output.len(),
+            "Updating state with calldata"
         );
         let snos_output = slice_slice_u8_to_vec_field(snos_output.as_slice());
         let layout_bridge_output = slice_slice_u8_to_vec_field(program_output.as_slice());
@@ -136,13 +138,16 @@ impl SettlementClient for StarknetSettlementClient {
             .update_state(snos_output, layout_bridge_output)
             .await
             .map_err(|e| eyre!("Failed to update state with calldata: {:?}", e))?;
-        tracing::info!(
+
+        let tx_hash = invoke_result.transaction_hash.to_hex_string();
+        info!(
             log_type = "completed",
             category = "update_state",
             function_type = "calldata",
-            "State updated with calldata."
+            tx_hash = %tx_hash,
+            "State update transaction submitted to Starknet with calldata"
         );
-        Ok(invoke_result.transaction_hash.to_hex_string())
+        Ok(tx_hash)
     }
 
     /// Should verify the inclusion of a tx in the settlement layer

orchestrator/src/core/client/alert/mod.rs

@@ -17,4 +17,14 @@ pub trait AlertClient: Send + Sync {
     ///
     /// # Returns
     async fn send_message(&self, message_body: String) -> Result<(), AlertError>;
+
+    /// Perform a health check on the alert service
+    ///
+    /// This method verifies that the alert service (e.g., AWS SNS) is accessible
+    /// and the necessary permissions are in place.
+    ///
+    /// # Returns
+    /// * `Ok(())` - If the alert service is healthy and accessible
+    /// * `Err(AlertError)` - If the health check fails
+    async fn health_check(&self) -> Result<(), AlertError>;
 }

orchestrator/src/core/client/alert/sns.rs

@@ -147,4 +147,19 @@ impl AlertClient for SNS {
         self.client().publish().topic_arn(self.get_topic_arn().await?).message(message_body).send().await?;
         Ok(())
     }
+
+    async fn health_check(&self) -> Result<(), AlertError> {
+        // Verify SNS accessibility by getting topic attributes
+        // This checks both connectivity and permissions
+        let topic_arn = self.get_topic_arn().await?;
+
+        self.client()
+            .get_topic_attributes()
+            .topic_arn(topic_arn)
+            .send()
+            .await
+            .map_err(|e| AlertError::TopicARNInvalid(format!("SNS health check failed: {}", e)))?;
+
+        Ok(())
+    }
 }

orchestrator/src/core/client/database/mod.rs

@@ -388,4 +388,19 @@ pub trait DatabaseClient: Send + Sync {
         &self,
         aggregator_index: u64,
     ) -> Result<Vec<SnosBatch>, DatabaseError>;
+
+    // ================================================================================
+    // Health Check Methods
+    // ================================================================================
+
+    /// Perform a health check on the database connection
+    ///
+    /// This method verifies that the database is accessible and operational.
+    /// It should perform a lightweight operation that validates connectivity
+    /// and basic functionality without impacting performance.
+    ///
+    /// # Returns
+    /// * `Ok(())` - If the database is healthy and accessible
+    /// * `Err(DatabaseError)` - If the health check fails
+    async fn health_check(&self) -> Result<(), DatabaseError>;
 }

orchestrator/src/core/client/database/mongodb.rs

@@ -1024,7 +1024,6 @@ impl DatabaseClient for MongoDbClient {
         Ok(batches)
     }
 
-    #[tracing::instrument(skip(self), fields(function_type = "db_call"), ret, err)]
     async fn get_snos_batches_without_jobs(
         &self,
         snos_batch_status: SnosBatchStatus,
@@ -1368,6 +1367,19 @@ impl DatabaseClient for MongoDbClient {
         ORCHESTRATOR_METRICS.db_calls_response_time.record(duration.as_secs_f64(), &attributes);
         Ok(updated_batches)
     }
+
+    async fn health_check(&self) -> Result<(), DatabaseError> {
+        let start = Instant::now();
+
+        // Perform a simple ping operation to verify connectivity
+        // This is a lightweight operation that checks if the database is accessible
+        self.database.run_command(doc! { "ping": 1 }, None).await?;
+
+        let attributes = [KeyValue::new("db_operation_name", "health_check")];
+        let duration = start.elapsed();
+        ORCHESTRATOR_METRICS.db_calls_response_time.record(duration.as_secs_f64(), &attributes);
+        Ok(())
+    }
 }
 
 // Generic utility function to convert Vec<T> to Option<T>

orchestrator/src/core/client/queue/mod.rs

@@ -16,4 +16,14 @@ pub trait QueueClient: Send + Sync {
     async fn get_consumer(&self, queue: QueueType) -> Result<SqsConsumer, QueueError>;
     async fn send_message(&self, queue: QueueType, payload: String, delay: Option<Duration>) -> Result<(), QueueError>;
     async fn consume_message_from_queue(&self, queue: QueueType) -> Result<Delivery, QueueError>;
+
+    /// Perform a health check on the queue service
+    ///
+    /// This method verifies that the queue service (e.g., AWS SQS) is accessible
+    /// and the necessary permissions are in place.
+    ///
+    /// # Returns
+    /// * `Ok(())` - If the queue service is healthy and accessible
+    /// * `Err(QueueError)` - If the health check fails
+    async fn health_check(&self) -> Result<(), QueueError>;
 }

orchestrator/src/core/client/queue/sqs.rs

@@ -311,7 +311,6 @@ impl QueueClient for SQS {
             .queue_url(&queue_url)
             .max_number_of_messages(1)
             .message_attribute_names("All")
-            .visibility_timeout(30)
             .send()
             .await?;
 
@@ -435,4 +434,15 @@ impl QueueClient for SQS {
 
         Ok(delivery)
     }
+
+    async fn health_check(&self) -> Result<(), QueueError> {
+        // Verify SQS accessibility by getting queue attributes
+        // This checks both connectivity and permissions
+        let queue_name = self.get_queue_name(&QueueType::WorkerTrigger)?;
+        let queue_url = self.inner.get_queue_url_from_client(queue_name.as_str()).await?;
+
+        self.inner.client().get_queue_attributes().queue_url(&queue_url).send().await?;
+
+        Ok(())
+    }
 }

orchestrator/src/core/client/storage/mod.rs

@@ -19,4 +19,14 @@ pub trait StorageClient: Send + Sync {
 
     /// List files in a directory
     async fn list_files_in_dir(&self, dir_path: &str) -> Result<Vec<String>, StorageError>;
+
+    /// Perform a health check on the storage service
+    ///
+    /// This method verifies that the storage service (e.g., AWS S3) is accessible
+    /// and the necessary permissions are in place.
+    ///
+    /// # Returns
+    /// * `Ok(())` - If the storage service is healthy and accessible
+    /// * `Err(StorageError)` - If the health check fails
+    async fn health_check(&self) -> Result<(), StorageError>;
 }