[bridge indexer] retry getting block number and add more logs when it err

crates/sui-bridge-indexer/src/eth_bridge_indexer.rs

@@ -117,8 +117,14 @@ impl Datasource<RawEthData> for EthSubscriptionDatasource {
                         }
                     }
                     _ = interval.tick() => {
-                        let latest_block = eth_ws_client.get_block_number().await?.as_u64();
-                        progress_metric.set(latest_block as i64);
+                        let Ok(Ok(block_num)) = retry_with_max_elapsed_time!(
+                            eth_ws_client.get_block_number(),
+                            Duration::from_secs(30000)
+                        ) else {
+                            tracing::error!("Failed to get block number");
+                            continue;
+                        };
+                        progress_metric.set(block_num.as_u64() as i64);
                     }
                 }
             }

crates/sui-indexer-builder/src/indexer_builder.rs

@@ -422,7 +422,13 @@ pub trait Datasource<T: Send>: Sync + Send {
         let mut stream = mysten_metrics::metered_channel::ReceiverStream::new(data_rx)
             .ready_chunks(ingestion_batch_size);
         let mut last_saved_checkpoint = None;
-        while let Some(batch) = stream.next().await {
+        loop {
+            let batch_option = stream.next().await;
+            if batch_option.is_none() {
+                tracing::error!(task_name, "Data stream ended unexpectedly");
+                break;
+            }
+            let batch = batch_option.unwrap();
             let mut max_height = 0;
             let mut heights = vec![];
             let mut data = vec![];
@@ -521,7 +527,9 @@ pub trait Datasource<T: Send>: Sync + Send {
         if let Some(m) = &remaining_checkpoints_metric {
             m.set(0)
         }
-        join_handle.await?
+        join_handle.await?.tap_err(|err| {
+            tracing::error!(task_name, "Data retrieval task failed: {:?}", err);
+        })
     }
 
     async fn start_data_retrieval(