feat: database support retry

yiweichi · yiweichi · commit 9d656092c303 · 2025-09-22T02:55:32.000+08:00
diff --git a/crates/chain-orchestrator/src/lib.rs b/crates/chain-orchestrator/src/lib.rs
@@ -18,8 +18,7 @@ use scroll_alloy_consensus::TxL1Message;
 use scroll_alloy_hardforks::ScrollHardforks;
 use scroll_alloy_network::Scroll;
 use scroll_db::{
-    Database, DatabaseError, DatabaseReadOperations, DatabaseTransactionProvider,
-    DatabaseWriteOperations, L1MessageStart, UnwindResult,
+    retry_with_defaults, Database, DatabaseError, DatabaseReadOperations, DatabaseTransactionProvider, DatabaseWriteOperations, L1MessageStart, UnwindResult
 };
 use scroll_network::NewBlockWithPeer;
 use std::{
@@ -211,9 +210,12 @@ impl<
         let database = ctx.database.clone();
         let block_info: L2BlockInfoWithL1Messages = (&block_with_peer.block).into();
         Self::do_handle_block_from_peer(ctx, block_with_peer).await?;
-        let tx = database.tx_mut().await?;
-        tx.update_l1_messages_with_l2_block(block_info.clone()).await?;
-        tx.commit().await?;
+        retry_with_defaults("update_l1_messages_with_l2_block", || async {
+            let tx = database.tx_mut().await?;
+            tx.update_l1_messages_with_l2_block(block_info.clone()).await?;
+            tx.commit().await?;
+            Ok::<_, DatabaseError>(())
+        }).await?;
         Ok(ChainOrchestratorEvent::L2ChainCommitted(block_info, None, true))
     }
 
@@ -292,10 +294,12 @@ impl<
         let mut received_chain_headers = VecDeque::from(vec![received_block.header.clone()]);
 
         // We should never have a re-org that is deeper than the current safe head.
-        let tx = database.tx().await?;
-        let (latest_safe_block, _) =
-            tx.get_latest_safe_l2_info().await?.expect("safe block must exist");
-        drop(tx);
+        let (latest_safe_block, _) = retry_with_defaults("get_latest_safe_l2_info", || async {     
+            let tx = database.tx().await?;
+            let (latest_safe_block, batch_info) =
+                tx.get_latest_safe_l2_info().await?.expect("safe block must exist");
+                Ok::<_, DatabaseError>((latest_safe_block, batch_info))
+        }).await?;
 
         // We search for the re-org index in the in-memory chain.
         const BATCH_FETCH_SIZE: usize = 50;
@@ -455,11 +459,14 @@ impl<
             ChainOrchestratorItem::InsertConsolidatedL2Blocks,
             Box::pin(async move {
                 let head = block_infos.last().expect("block info must not be empty").clone();
-                let tx = database.tx_mut().await?;
-                for block in block_infos {
-                    tx.insert_block(block, batch_info).await?;
-                }
-                tx.commit().await?;
+                retry_with_defaults("insert_block", || async {
+                    let tx = database.tx_mut().await?;
+                    for block in block_infos.clone() {
+                        tx.insert_block(block, batch_info).await?;
+                    }
+                    tx.commit().await?;
+                    Ok::<_, DatabaseError>(())
+                }).await?;
                 Result::<_, ChainOrchestratorError>::Ok(Some(
                     ChainOrchestratorEvent::L2ConsolidatedBlockCommitted(head),
                 ))
@@ -503,9 +510,12 @@ impl<
 
                 // Insert the blocks into the database.
                 let head = block_info.last().expect("block info must not be empty").clone();
-                let tx = database.tx_mut().await?;
-                tx.update_l1_messages_from_l2_blocks(block_info).await?;
-                tx.commit().await?;
+                retry_with_defaults("update_l1_messages_from_l2_blocks", || async {
+                    let tx = database.tx_mut().await?;
+                    tx.update_l1_messages_from_l2_blocks(block_info.clone()).await?;
+                    tx.commit().await?;
+                    Ok::<_, DatabaseError>(())
+                }).await?;
 
                 Result::<_, ChainOrchestratorError>::Ok(Some(
                     ChainOrchestratorEvent::L2ChainCommitted(head, None, consolidated),
@@ -589,10 +599,13 @@ impl<
         l2_client: Arc<P>,
         current_chain: Arc<Mutex<Chain>>,
     ) -> Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError> {
-        let txn = database.tx_mut().await?;
-        let UnwindResult { l1_block_number, queue_index, l2_head_block_number, l2_safe_block_info } =
-            txn.unwind(chain_spec.genesis_hash(), l1_block_number).await?;
-        txn.commit().await?;
+        let UnwindResult { l1_block_number, queue_index, l2_head_block_number, l2_safe_block_info } = retry_with_defaults("unwind", || async {
+            let txn = database.tx_mut().await?;
+            let UnwindResult { l1_block_number, queue_index, l2_head_block_number, l2_safe_block_info } =
+                txn.unwind(chain_spec.genesis_hash(), l1_block_number).await?;
+            txn.commit().await?;
+            Ok::<_, DatabaseError>(UnwindResult { l1_block_number, queue_index, l2_head_block_number, l2_safe_block_info })
+        }).await?;
         let l2_head_block_info = if let Some(block_number) = l2_head_block_number {
             // Fetch the block hash of the new L2 head block.
             let block_hash = l2_client
@@ -623,16 +636,21 @@ impl<
         block_number: u64,
         l1_block_number: Arc<AtomicU64>,
     ) -> Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError> {
-        let tx = database.tx_mut().await?;
+        let finalized_batches = retry_with_defaults("set_latest_finalized_l1_block_number", || async {
+            let tx = database.tx_mut().await?;
 
-        // Set the latest finalized L1 block in the database.
-        tx.set_latest_finalized_l1_block_number(block_number).await?;
+            // Set the latest finalized L1 block in the database.
+            tx.set_latest_finalized_l1_block_number(block_number).await?;
+    
+            // Get all unprocessed batches that have been finalized by this L1 block finalization.
+            let finalized_batches =
+                tx.fetch_and_update_unprocessed_finalized_batches(block_number).await?;
+    
+            tx.commit().await?;
 
-        // Get all unprocessed batches that have been finalized by this L1 block finalization.
-        let finalized_batches =
-            tx.fetch_and_update_unprocessed_finalized_batches(block_number).await?;
+            Ok::<_, DatabaseError>(finalized_batches)
+        }).await?;
 
-        tx.commit().await?;
 
         // Update the chain orchestrator L1 block number.
         l1_block_number.store(block_number, Ordering::Relaxed);
@@ -654,17 +672,20 @@ impl<
         let l1_message = L1MessageEnvelope::new(l1_message, l1_block_number, None, queue_hash);
 
         // Perform a consistency check to ensure the previous L1 message exists in the database.
-        let tx = database.tx_mut().await?;
-        if l1_message.transaction.queue_index > 0 &&
-            tx.get_l1_message_by_index(l1_message.transaction.queue_index - 1).await?.is_none()
-        {
-            return Err(ChainOrchestratorError::L1MessageQueueGap(
-                l1_message.transaction.queue_index,
-            ))
-        }
+        let _ = retry_with_defaults("handle_l1_message", || async {
+            let tx = database.tx_mut().await?;
+            if l1_message.transaction.queue_index > 0 &&
+                tx.get_l1_message_by_index(l1_message.transaction.queue_index - 1).await?.is_none()
+            {
+                return Err(ChainOrchestratorError::L1MessageQueueGap(
+                    l1_message.transaction.queue_index,
+                ))
+            }
 
-        tx.insert_l1_message(l1_message).await?;
-        tx.commit().await?;
+            tx.insert_l1_message(l1_message.clone()).await?;
+            tx.commit().await?;
+            Ok::<_, ChainOrchestratorError>(())
+        }).await;
         Ok(Some(event))
     }
 
@@ -673,36 +694,41 @@ impl<
         database: Arc<Database>,
         batch: BatchCommitData,
     ) -> Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError> {
-        let tx = database.tx_mut().await?;
-        let prev_batch_index = batch.index - 1;
-
-        // Perform a consistency check to ensure the previous commit batch exists in the database.
-        if tx.get_batch_by_index(prev_batch_index).await?.is_none() {
-            return Err(ChainOrchestratorError::BatchCommitGap(batch.index))
-        }
-
-        // remove any batches with an index greater than the previous batch.
-        let affected = tx.delete_batches_gt_batch_index(prev_batch_index).await?;
-
-        // handle the case of a batch revert.
-        let new_safe_head = if affected > 0 {
-            tx.delete_l2_blocks_gt_batch_index(prev_batch_index).await?;
-            tx.get_highest_block_for_batch_index(prev_batch_index).await?
-        } else {
-            None
-        };
-
-        let event = ChainOrchestratorEvent::BatchCommitIndexed {
-            batch_info: BatchInfo::new(batch.index, batch.hash),
-            l1_block_number: batch.block_number,
-            safe_head: new_safe_head,
-        };
+        let event = retry_with_defaults("handle_batch_commit", || async {
+            let tx = database.tx_mut().await?;
+            let batch_clone = batch.clone();
+            let prev_batch_index = batch_clone.clone().index - 1;
+            
+            // Perform a consistency check to ensure the previous commit batch exists in the database.
+            if tx.get_batch_by_index(prev_batch_index).await?.is_none() {
+                return Err(ChainOrchestratorError::BatchCommitGap(batch_clone.index))
+            }
+    
+            // remove any batches with an index greater than the previous batch.
+            let affected = tx.delete_batches_gt_batch_index(prev_batch_index).await?;
+    
+            // handle the case of a batch revert.
+            let new_safe_head = if affected > 0 {
+                tx.delete_l2_blocks_gt_batch_index(prev_batch_index).await?;
+                tx.get_highest_block_for_batch_index(prev_batch_index).await?
+            } else {
+                None
+            };
+    
+            let event = ChainOrchestratorEvent::BatchCommitIndexed {
+                batch_info: BatchInfo::new(batch_clone.index, batch_clone.hash),
+                l1_block_number: batch.block_number,
+                safe_head: new_safe_head,
+            };
+    
+            // insert the batch and commit the transaction.
+            tx.insert_batch(batch_clone).await?;
+            tx.commit().await?;
 
-        // insert the batch and commit the transaction.
-        tx.insert_batch(batch).await?;
-        tx.commit().await?;
+            Ok::<_, ChainOrchestratorError>(Some(event))
+        }).await?;
 
-        Ok(Some(event))
+        Ok(event)
     }
 
     /// Handles a batch finalization event by updating the batch input in the database.
@@ -712,22 +738,24 @@ impl<
         block_number: u64,
         finalized_block_number: Arc<AtomicU64>,
     ) -> Result<Option<ChainOrchestratorEvent>, ChainOrchestratorError> {
-        let tx = database.tx_mut().await?;
+        retry_with_defaults("handle_batch_finalization", || async {
+            let tx = database.tx_mut().await?;
 
-        // finalize all batches up to `batch_index`.
-        tx.finalize_batches_up_to_index(batch_index, block_number).await?;
+            // finalize all batches up to `batch_index`.
+            tx.finalize_batches_up_to_index(batch_index, block_number).await?;
 
-        // Get all unprocessed batches that have been finalized by this L1 block finalization.
-        let finalized_block_number = finalized_block_number.load(Ordering::Relaxed);
-        if finalized_block_number >= block_number {
-            let finalized_batches =
-                tx.fetch_and_update_unprocessed_finalized_batches(finalized_block_number).await?;
-            tx.commit().await?;
-            return Ok(Some(ChainOrchestratorEvent::BatchFinalized(block_number, finalized_batches)))
-        }
+            // Get all unprocessed batches that have been finalized by this L1 block finalization.
+            let finalized_block_number = finalized_block_number.load(Ordering::Relaxed);
+            if finalized_block_number >= block_number {
+                let finalized_batches =
+                    tx.fetch_and_update_unprocessed_finalized_batches(finalized_block_number).await?;
+                tx.commit().await?;
+                return Ok(Some(ChainOrchestratorEvent::BatchFinalized(block_number, finalized_batches)))
+            }
 
-        tx.commit().await?;
-        Ok(None)
+            tx.commit().await?;
+            Ok::<_, ChainOrchestratorError>(None)
+        }).await
     }
 }
 
@@ -772,8 +800,8 @@ async fn init_chain_from_db<P: Provider<Scroll> + 'static>(
 ) -> Result<BoundedVec<Header>, ChainOrchestratorError> {
     let blocks = {
         let mut blocks = Vec::with_capacity(chain_buffer_size);
-        let tx = database.tx().await?;
-        let blocks_stream = tx.get_l2_blocks().await?.take(chain_buffer_size);
+        let tx = retry_with_defaults("get_l2_blocks_new_tx", || database.tx()).await?;
+        let blocks_stream = retry_with_defaults("get_l2_blocks", || tx.get_l2_blocks()).await?.take(chain_buffer_size);
         pin_mut!(blocks_stream);
         while let Some(block_info) = blocks_stream.try_next().await? {
             let header = l2_client
@@ -869,9 +897,11 @@ async fn consolidate_chain<P: Provider<Scroll> + 'static>(
 
     // Fetch the safe head from the database. We use this as a trust anchor to reconcile the chain
     // back to.
-    let tx = database.tx().await?;
-    let safe_head = tx.get_latest_safe_l2_info().await?.expect("safe head must exist").0;
-    drop(tx);
+    let safe_head = retry_with_defaults("get_latest_safe_l2_info", || async {
+        let tx = database.tx().await?;
+        let safe_head = tx.get_latest_safe_l2_info().await?.expect("safe head must exist").0;
+        Ok::<_, DatabaseError>(safe_head)
+    }).await?;
 
     // If the in-memory chain contains the safe head, we check if the safe hash from the
     // database (L1 consolidation) matches the in-memory value. If it does not match, we return an
@@ -970,9 +1000,8 @@ async fn validate_l1_messages(
     // TODO: instead of using `l1_message_hashes.first().map(|tx| L1MessageStart::Hash(*tx))` to
     // determine the start of the L1 message stream, we should use a more robust method to determine
     // the start of the L1 message stream.
-    let tx = database.tx().await?;
-    let l1_message_stream =
-        tx.get_l1_messages(l1_message_hashes.first().map(|tx| L1MessageStart::Hash(*tx))).await?;
+    let tx = retry_with_defaults("get_l1_messages_new_tx", || database.tx()).await?;
+    let l1_message_stream = retry_with_defaults("get_l1_messages", || tx.get_l1_messages(l1_message_hashes.first().map(|tx| L1MessageStart::Hash(*tx)))).await?;
     pin_mut!(l1_message_stream);
 
     for message_hash in l1_message_hashes {
diff --git a/crates/database/db/src/db.rs b/crates/database/db/src/db.rs
@@ -136,7 +136,7 @@ impl From<DatabaseConnection> for Database {
     }
 }
 
-#[cfg(test)]
+#[cfg(all(test, feature = "test-utils"))]
 mod test {
     use super::*;
     use crate::{
diff --git a/crates/database/db/src/lib.rs b/crates/database/db/src/lib.rs
@@ -19,6 +19,9 @@ pub use operations::{
     DatabaseReadOperations, DatabaseWriteOperations, L1MessageStart, UnwindResult,
 };
 
+mod retry;
+pub use retry::{RetryConfig, retry_config, retry_operation_with_name, retry_with_defaults};
+
 mod transaction;
 pub use transaction::{DatabaseTransactionProvider, TXMut, TX};
 
diff --git a/crates/database/db/src/retry.rs b/crates/database/db/src/retry.rs

Original file line number	Diff line number	Diff line change
`@@ -136,7 +136,7 @@ impl From<DatabaseConnection> for Database {`
`136`	`136`	`}`
`137`	`137`	`}`
`138`	`138`
`139`		`-#[cfg(test)]`
	`139`	`+#[cfg(all(test, feature = "test-utils"))]`
`140`	`140`	`mod test {`
`141`	`141`	`use super::*;`
`142`	`142`	`use crate::{`