lint

rust/processor/src/processors/account_transactions_processor.rs

@@ -13,7 +13,6 @@ use crate::{
 use anyhow::bail;
 use aptos_protos::transaction::v1::Transaction;
 use async_trait::async_trait;
-use diesel::result::Error;
 use field_count::FieldCount;
 use std::{collections::HashMap, fmt::Debug};
 use tracing::error;

rust/processor/src/processors/ans_processor.rs

@@ -24,7 +24,7 @@ use aptos_protos::transaction::v1::{
     transaction::TxnData, write_set_change::Change as WriteSetChange, Transaction,
 };
 use async_trait::async_trait;
-use diesel::{pg::upsert::excluded, result::Error, ExpressionMethods};
+use diesel::{pg::upsert::excluded, ExpressionMethods};
 use field_count::FieldCount;
 use serde::{Deserialize, Serialize};
 use std::{collections::HashMap, fmt::Debug};

rust/processor/src/processors/coin_processor.rs

@@ -21,7 +21,7 @@ use crate::{
 use anyhow::bail;
 use aptos_protos::transaction::v1::Transaction;
 use async_trait::async_trait;
-use diesel::{pg::upsert::excluded, result::Error, ExpressionMethods};
+use diesel::{pg::upsert::excluded, ExpressionMethods};
 use field_count::FieldCount;
 use std::{collections::HashMap, fmt::Debug};
 use tracing::error;
@@ -92,27 +92,20 @@ async fn insert_to_db(
     {
         Ok(_) => Ok(()),
         Err(_) => {
-            conn.build_transaction()
-                .read_write()
-                .run::<_, Error, _>(|pg_conn| {
-                    Box::pin(async {
-                        let coin_activities = clean_data_for_db(coin_activities, true);
-                        let coin_infos = clean_data_for_db(coin_infos, true);
-                        let coin_balances = clean_data_for_db(coin_balances, true);
-                        let current_coin_balances = clean_data_for_db(current_coin_balances, true);
-                        let coin_supply = clean_data_for_db(coin_supply, true);
+            let coin_activities = clean_data_for_db(coin_activities, true);
+            let coin_infos = clean_data_for_db(coin_infos, true);
+            let coin_balances = clean_data_for_db(coin_balances, true);
+            let current_coin_balances = clean_data_for_db(current_coin_balances, true);
+            let coin_supply = clean_data_for_db(coin_supply, true);
 
-                        insert_to_db_impl(
-                            pg_conn,
-                            &coin_activities,
-                            &coin_infos,
-                            &coin_balances,
-                            &current_coin_balances,
-                            &coin_supply,
-                        )
-                        .await
-                    })
-                })
+            insert_to_db_impl(
+                conn,
+                &coin_activities,
+                &coin_infos,
+                &coin_balances,
+                &current_coin_balances,
+                &coin_supply,
+            )
                 .await
         },
     }

rust/processor/src/processors/default_processor.rs

@@ -20,7 +20,7 @@ use crate::{
 use anyhow::bail;
 use aptos_protos::transaction::v1::Transaction;
 use async_trait::async_trait;
-use diesel::{pg::upsert::excluded, result::Error, ExpressionMethods};
+use diesel::{pg::upsert::excluded, ExpressionMethods};
 use field_count::FieldCount;
 use std::{collections::HashMap, fmt::Debug};
 use tracing::error;

rust/processor/src/processors/events_processor.rs

@@ -13,7 +13,7 @@ use crate::{
 use anyhow::bail;
 use aptos_protos::transaction::v1::{transaction::TxnData, Transaction};
 use async_trait::async_trait;
-use diesel::{pg::upsert::excluded, result::Error, ExpressionMethods};
+use diesel::{pg::upsert::excluded, ExpressionMethods};
 use field_count::FieldCount;
 use std::fmt::Debug;
 use tracing::error;

rust/processor/src/processors/fungible_asset_processor.rs

@@ -30,7 +30,7 @@ use anyhow::bail;
 use aptos_protos::transaction::v1::{transaction::TxnData, write_set_change::Change, Transaction};
 use async_trait::async_trait;
 use chrono::NaiveDateTime;
-use diesel::{pg::upsert::excluded, result::Error, ExpressionMethods};
+use diesel::{pg::upsert::excluded, ExpressionMethods};
 use field_count::FieldCount;
 use std::{collections::HashMap, fmt::Debug};
 use tracing::error;

rust/processor/src/processors/monitoring_processor.rs

@@ -21,7 +21,7 @@ use crate::{
 use anyhow::bail;
 use aptos_protos::transaction::v1::{write_set_change::Change, Transaction};
 use async_trait::async_trait;
-use diesel::{pg::upsert::excluded, result::Error, ExpressionMethods};
+use diesel::{pg::upsert::excluded, ExpressionMethods};
 use field_count::FieldCount;
 use std::{collections::HashMap, fmt::Debug};
 use tracing::error;

rust/processor/src/processors/objects_processor.rs

@@ -25,7 +25,7 @@ use crate::{
 use anyhow::bail;
 use aptos_protos::transaction::v1::{write_set_change::Change, Transaction};
 use async_trait::async_trait;
-use diesel::{pg::upsert::excluded, result::Error, ExpressionMethods};
+use diesel::{pg::upsert::excluded, ExpressionMethods};
 use field_count::FieldCount;
 use std::{collections::HashMap, fmt::Debug};
 use tracing::error;

rust/processor/src/processors/stake_processor.rs

@@ -28,7 +28,7 @@ use crate::{
 use anyhow::bail;
 use aptos_protos::transaction::v1::{write_set_change::Change, Transaction};
 use async_trait::async_trait;
-use diesel::{pg::upsert::excluded, result::Error, ExpressionMethods};
+use diesel::{pg::upsert::excluded, ExpressionMethods};
 use field_count::FieldCount;
 use std::{collections::HashMap, fmt::Debug};
 use tracing::error;

rust/processor/src/processors/token_processor.rs

@@ -24,7 +24,7 @@ use crate::{
 use anyhow::bail;
 use aptos_protos::transaction::v1::Transaction;
 use async_trait::async_trait;
-use diesel::{pg::upsert::excluded, result::Error, ExpressionMethods};
+use diesel::{pg::upsert::excluded, ExpressionMethods};
 use field_count::FieldCount;
 use serde::{Deserialize, Serialize};
 use std::{collections::HashMap, fmt::Debug};

rust/processor/src/processors/token_v2_processor.rs

@@ -38,7 +38,7 @@ use crate::{
 use anyhow::bail;
 use aptos_protos::transaction::v1::{transaction::TxnData, write_set_change::Change, Transaction};
 use async_trait::async_trait;
-use diesel::{pg::upsert::excluded, result::Error, ExpressionMethods};
+use diesel::{pg::upsert::excluded, ExpressionMethods};
 use field_count::FieldCount;
 use std::{
     collections::{HashMap, HashSet},

rust/processor/src/processors/user_transaction_processor.rs

@@ -15,7 +15,7 @@ use crate::{
 use anyhow::bail;
 use aptos_protos::transaction::v1::{transaction::TxnData, Transaction};
 use async_trait::async_trait;
-use diesel::{pg::upsert::excluded, result::Error, ExpressionMethods};
+use diesel::{pg::upsert::excluded, ExpressionMethods};
 use field_count::FieldCount;
 use std::fmt::Debug;
 use tracing::error;

rust/processor/src/worker.rs

@@ -38,7 +38,6 @@ use futures::StreamExt;
 use prost::Message;
 use std::{sync::Arc, time::Duration};
 use diesel::Connection;
-use futures_util::TryFutureExt;
 use tokio::{sync::mpsc::error::TryRecvError, time::timeout};
 use tonic::{Response, Streaming};
 use tracing::{error, info};