chore: improve slot persister performance, clean up deps

Author: armyhaylenko

Cargo.lock

@@ -3,7 +3,7 @@ use std::{sync::Arc, time::Duration};
 use async_trait::async_trait;
 use entities::{
     models::{RawBlockWithTransactions, TransactionInfo},
-    utils::decode_encoded_transaction_with_status_meta,
+    transaction_converters::decode_encoded_transaction_with_status_meta,
 };
 use interface::{error::StorageError, signature_persistence::BlockProducer};
 use solana_client::rpc_config::RpcBlockConfig;

backfill_rpc/src/block_producer.rs

@@ -2,4 +2,4 @@ pub mod api_req_params;
 pub mod enums;
 pub mod models;
 pub mod schedule;
-pub mod utils;
+pub mod transaction_converters;

entities/src/lib.rs

@@ -451,13 +451,14 @@ impl From<RawBlockDeprecated> for RawBlock {
                     .transactions
                     .into_iter()
                     .filter_map(|t| {
-                        crate::utils::decode_encoded_transaction_with_status_meta(t).and_then(
-                            |tx| {
-                                TransactionInfo::from_transaction_with_status_meta_and_slot(
-                                    tx, value.slot,
-                                )
-                            },
+                        crate::transaction_converters::decode_encoded_transaction_with_status_meta(
+                            t,
                         )
+                        .and_then(|tx| {
+                            TransactionInfo::from_transaction_with_status_meta_and_slot(
+                                tx, value.slot,
+                            )
+                        })
                     })
                     .collect(),
             },

entities/src/models.rs

@@ -8,7 +8,7 @@ use solana_sdk::{
 use solana_transaction_status::{
     option_serializer::OptionSerializer, EncodedTransactionWithStatusMeta, InnerInstruction,
     InnerInstructions, TransactionStatusMeta, TransactionTokenBalance, TransactionWithStatusMeta,
-    UiInstruction, VersionedTransactionWithStatusMeta,
+    UiInstruction, UiTransactionTokenBalance, VersionedTransactionWithStatusMeta,
 };
 
 pub fn decode_encoded_transaction_with_status_meta(
@@ -51,28 +51,10 @@ pub fn decode_encoded_transaction_with_status_meta(
                 post_balances: m.post_balances,
                 inner_instructions,
                 log_messages: get(m.log_messages),
-                pre_token_balances: get(m.pre_token_balances).map(|uttb| {
-                    uttb.into_iter()
-                        .map(|uttb| TransactionTokenBalance {
-                            account_index: uttb.account_index,
-                            mint: uttb.mint,
-                            ui_token_amount: uttb.ui_token_amount,
-                            owner: get(uttb.owner).unwrap_or_default(),
-                            program_id: get(uttb.program_id).unwrap_or_default(),
-                        })
-                        .collect()
-                }),
-                post_token_balances: get(m.post_token_balances).map(|uttb| {
-                    uttb.into_iter()
-                        .map(|uttb| TransactionTokenBalance {
-                            account_index: uttb.account_index,
-                            mint: uttb.mint,
-                            ui_token_amount: uttb.ui_token_amount,
-                            owner: Option::<String>::from(uttb.owner).unwrap_or_default(),
-                            program_id: Option::<String>::from(uttb.program_id).unwrap_or_default(),
-                        })
-                        .collect()
-                }),
+                pre_token_balances: get(m.pre_token_balances)
+                    .map(decode_transaction_token_balances),
+                post_token_balances: get(m.post_token_balances)
+                    .map(decode_transaction_token_balances),
                 rewards: m.rewards.into(),
                 loaded_addresses: get(m.loaded_addresses)
                     .map(|ula| LoadedAddresses {
@@ -105,3 +87,18 @@ pub fn decode_encoded_transaction_with_status_meta(
         None => None,
     }
 }
+
+fn decode_transaction_token_balances(
+    ui_transaction_token_balances: Vec<UiTransactionTokenBalance>,
+) -> Vec<TransactionTokenBalance> {
+    ui_transaction_token_balances
+        .into_iter()
+        .map(|uttb| TransactionTokenBalance {
+            account_index: uttb.account_index,
+            mint: uttb.mint,
+            ui_token_amount: uttb.ui_token_amount,
+            owner: Option::<String>::from(uttb.owner).unwrap_or_default(),
+            program_id: Option::<String>::from(uttb.program_id).unwrap_or_default(),
+        })
+        .collect()
+}

entities/src/utils.rs entities/src/transaction_converters.rs

@@ -18,7 +18,6 @@ async-trait = { workspace = true }
 thiserror = { workspace = true }
 solana-transaction-status = { workspace = true }
 tracing = { workspace = true }
-rocks-db = { path = "../rocks-db" }
 
 [dev-dependencies]
 mockall = { workspace = true }

grpc/Cargo.toml

@@ -11,7 +11,6 @@ use interface::{
     error::StorageError,
     signature_persistence::BlockProducer,
 };
-use rocks_db::column::TypedColumn;
 use tonic::{
     transport::{Channel, Uri},
     Code, Status,
@@ -104,7 +103,7 @@ impl BlockProducer for Client {
             .await
             .map_err(|e| StorageError::Common(e.to_string()))
             .and_then(|response| {
-                RawBlock::decode(response.into_inner().block.as_slice())
+                bincode::deserialize::<RawBlock>(response.into_inner().block.as_slice())
                     .map_err(|e| StorageError::Common(e.to_string()))
                     .map(|raw_block| raw_block.block)
             })

grpc/src/client.rs

@@ -452,23 +452,19 @@ async fn process_slots(
                     successful_blocks.len()
                 );
 
-                match target_db.raw_blocks.put_batch(successful_blocks.clone()).await {
+                let projected_last_slot = successful_blocks.keys().max().copied().unwrap_or(0);
+                let successful_blocks_len = successful_blocks.len();
+                match target_db.raw_blocks.put_batch(std::mem::take(&mut successful_blocks)).await {
                     Ok(_) => {
-                        let last_slot = successful_blocks.keys().max().cloned().unwrap_or(0);
                         info!(
                             "Successfully saved {} blocks to RocksDB. Last stored slot: {}",
-                            successful_blocks.len(),
-                            last_slot
+                            successful_blocks_len, projected_last_slot
                         );
                         break; // Move on to next chunk of `slots`
                     },
                     Err(e) => {
                         // DB write failed
-                        error!(
-                            "Failed to save {} blocks to RocksDB: {}",
-                            successful_blocks.len(),
-                            e
-                        );
+                        error!("Failed to save {} blocks to RocksDB: {}", successful_blocks_len, e);
                         batch_retries += 1;
                         if batch_retries >= MAX_BATCH_RETRIES {
                             panic!(

nft_ingester/src/bin/slot_persister/main.rs

@@ -33,5 +33,4 @@ thiserror = { workspace = true }
 bubblegum-batch-sdk = { workspace = true }
 solana-storage-proto = { workspace = true }
 prost = "0.11.9"
-prost-derive = "0.11.9"
 futures = { workspace = true }