refactor(DB): rollback tx when failed rather than close connection

core/storage/src/relational/mod.rs

@@ -20,7 +20,7 @@ use common::{
     PaginationResponse, Range, Result,
 };
 use common_logger::{tracing, tracing_async};
-use db_xsql::{rbatis::Bytes as RbBytes, XSQLPool};
+use db_xsql::{commit_transaction, rbatis::Bytes as RbBytes, XSQLPool};
 use protocol::db::{DBDriver, DBInfo, SimpleBlock, SimpleTransaction, TransactionWrapper};
 
 use ckb_types::core::{BlockNumber, BlockView, HeaderView};
@@ -51,7 +51,7 @@ impl Storage for RelationalStorage {
         self.insert_transaction_table(ctx.clone(), &block, &mut tx)
             .await?;
 
-        tx.commit().await?;
+        commit_transaction(tx).await?;
         Ok(())
     }
 
@@ -69,7 +69,7 @@ impl Storage for RelationalStorage {
             .await?;
         self.remove_block_table(ctx.clone(), block_number, block_hash, &mut tx)
             .await?;
-        tx.commit().await?;
+        commit_transaction(tx).await?;
 
         Ok(())
     }
@@ -545,7 +545,7 @@ impl Storage for RelationalStorage {
         let res = self
             .insert_registered_address_table(addresses, &mut tx)
             .await?;
-        tx.commit().await?;
+        commit_transaction(tx).await?;
 
         Ok(res
             .iter()

core/synchronization/src/lib.rs

@@ -9,7 +9,7 @@ use core_storage::relational::table::{
     IO_TYPE_INPUT, IO_TYPE_OUTPUT,
 };
 use core_storage::relational::{generate_id, to_rb_bytes, BATCH_SIZE_THRESHOLD};
-use db_xsql::{rbatis::crud::CRUDMut, XSQLPool};
+use db_xsql::{commit_transaction, rbatis::crud::CRUDMut, XSQLPool};
 
 use ckb_types::core::{BlockNumber, BlockView};
 use ckb_types::prelude::*;
@@ -335,9 +335,8 @@ async fn sync_indexer_cell(task: Vec<BlockNumber>, rdb: XSQLPool) -> Result<()>
         .for_each(|c| c.id = generate_id(c.block_number));
     core_storage::save_batch_slice!(tx, indexer_cells, status_list);
 
-    tx.commit().await?;
+    commit_transaction(tx).await?;
 
-    let _ = tx.take_conn().unwrap().close().await;
     free_one_task();
 
     Ok(())
@@ -419,9 +418,7 @@ async fn sync_blocks<T: SyncAdapter>(
         canonical_data_table_batch
     );
 
-    tx.commit().await?;
-
-    let _ = tx.take_conn().unwrap().close().await;
+    commit_transaction(tx).await?;
 
     Ok(())
 }

db/xsql/src/lib.rs

@@ -2,7 +2,7 @@ pub mod page;
 
 pub use rbatis;
 
-use common::Result;
+use common::{anyhow::anyhow, Result};
 use protocol::db::DBDriver;
 
 use log::LevelFilter;
@@ -146,6 +146,15 @@ impl XSQLPool {
     }
 }
 
+pub async fn commit_transaction(mut tx: RBatisTxExecutor<'_>) -> Result<()> {
+    if tx.commit().await.is_err() {
+        tx.rollback().await?;
+        return Err(anyhow!("Commit transaction failed, transaction rollback!"));
+    }
+
+    Ok(())
+}
+
 fn build_url(
     db_type: &str,
     db_name: &str,

src/main.rs

@@ -1,6 +1,7 @@
 #[tokio::main(flavor = "multi_thread")]
 async fn main() {
     std::panic::set_hook(Box::new(move |info| {
+        println!("panic occurred {:?}", info);
         log::error!("panic occurred {:?}", info);
         std::process::exit(-1);
     }));