wrap producedblock struct and avoid cloning it (#336)

wrap ProducedBlock in Arc

blockstore/src/block_stores/postgres/postgres_block.rs

@@ -2,7 +2,7 @@ use super::postgres_epoch::PostgresEpoch;
 use super::postgres_session::PostgresSession;
 use log::{debug, warn};
 use solana_lite_rpc_core::structures::epoch::EpochRef;
-use solana_lite_rpc_core::structures::produced_block::TransactionInfo;
+use solana_lite_rpc_core::structures::produced_block::{ProducedBlockInner, TransactionInfo};
 use solana_lite_rpc_core::{encoding::BASE64, structures::produced_block::ProducedBlock};
 use solana_sdk::clock::Slot;
 use solana_sdk::commitment_config::CommitmentConfig;
@@ -56,7 +56,7 @@ impl PostgresBlock {
             .map(|x| BASE64.deserialize::<Vec<Reward>>(x).ok())
             .unwrap_or(None);
 
-        ProducedBlock {
+        let inner = ProducedBlockInner {
             // TODO implement
             transactions: transaction_infos,
             leader_id: None,
@@ -65,10 +65,10 @@ impl PostgresBlock {
             slot: self.slot as Slot,
             parent_slot: self.parent_slot as Slot,
             block_time: self.block_time as u64,
-            commitment_config,
             previous_blockhash: self.previous_blockhash.clone(),
             rewards: rewards_vec,
-        }
+        };
+        ProducedBlock::new(inner, commitment_config)
     }
 }
 

blockstore/src/block_stores/postgres/postgres_block_store_writer.rs

@@ -316,7 +316,7 @@ fn div_ceil(a: usize, b: usize) -> usize {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use solana_lite_rpc_core::structures::produced_block::TransactionInfo;
+    use solana_lite_rpc_core::structures::produced_block::{ProducedBlockInner, TransactionInfo};
     use solana_sdk::commitment_config::CommitmentConfig;
     use solana_sdk::signature::Signature;
     use std::str::FromStr;
@@ -364,7 +364,7 @@ mod tests {
         let sig1 = Signature::from_str("5VBroA4MxsbZdZmaSEb618WRRwhWYW9weKhh3md1asGRx7nXDVFLua9c98voeiWdBE7A9isEoLL7buKyaVRSK1pV").unwrap();
         let sig2 = Signature::from_str("3d9x3rkVQEoza37MLJqXyadeTbEJGUB6unywK4pjeRLJc16wPsgw3dxPryRWw3UaLcRyuxEp1AXKGECvroYxAEf2").unwrap();
 
-        ProducedBlock {
+        let inner = ProducedBlockInner {
             block_height: 42,
             blockhash: "blockhash".to_string(),
             previous_blockhash: "previous_blockhash".to_string(),
@@ -373,10 +373,10 @@ mod tests {
             transactions: vec![create_test_tx(sig1), create_test_tx(sig2)],
             // TODO double if this is unix millis or seconds
             block_time: 1699260872000,
-            commitment_config: CommitmentConfig::finalized(),
             leader_id: None,
             rewards: None,
-        }
+        };
+        ProducedBlock::new(inner, CommitmentConfig::finalized())
     }
 
     fn create_test_tx(signature: Signature) -> TransactionInfo {

blockstore/tests/blockstore_integration_tests.rs

@@ -296,15 +296,11 @@ fn spawn_client_to_blockstorage(
         // note: no startup deloy
         loop {
             match blocks_notifier.recv().await {
-                Ok(ProducedBlock {
-                    slot,
-                    commitment_config,
-                    ..
-                }) => {
-                    if commitment_config != CommitmentConfig::confirmed() {
+                Ok(produced_block) => {
+                    if produced_block.commitment_config != CommitmentConfig::confirmed() {
                         continue;
                     }
-                    let confirmed_slot = slot;
+                    let confirmed_slot = produced_block.slot;
                     // we cannot expect the most recent data
                     let query_slot = confirmed_slot - 3;
                     match block_storage_query.query_block(query_slot).await {

blockstore/tests/multiple_strategy_block_store_tests.rs

@@ -4,21 +4,20 @@ use solana_lite_rpc_blockstore::block_stores::postgres::postgres_block_store_que
 use solana_lite_rpc_blockstore::block_stores::postgres::postgres_block_store_writer::PostgresBlockStore;
 use solana_lite_rpc_blockstore::block_stores::postgres::PostgresSessionConfig;
 use solana_lite_rpc_core::structures::epoch::EpochCache;
-use solana_lite_rpc_core::structures::produced_block::ProducedBlock;
+use solana_lite_rpc_core::structures::produced_block::{ProducedBlock, ProducedBlockInner};
 use solana_sdk::pubkey::Pubkey;
 use solana_sdk::reward_type::RewardType;
 use solana_sdk::{commitment_config::CommitmentConfig, hash::Hash};
 use solana_transaction_status::Reward;
 
 pub fn create_test_block(slot: u64, commitment_config: CommitmentConfig) -> ProducedBlock {
-    ProducedBlock {
+    let inner = ProducedBlockInner {
         block_height: slot,
         blockhash: Hash::new_unique().to_string(),
         previous_blockhash: Hash::new_unique().to_string(),
         parent_slot: slot - 1,
         transactions: vec![],
         block_time: 0,
-        commitment_config,
         leader_id: None,
         slot,
         rewards: Some(vec![Reward {
@@ -28,7 +27,8 @@ pub fn create_test_block(slot: u64, commitment_config: CommitmentConfig) -> Prod
             reward_type: Some(RewardType::Voting),
             commission: None,
         }]),
-    }
+    };
+    ProducedBlock::new(inner, commitment_config)
 }
 
 #[ignore = "need postgres database"]

cluster-endpoints/src/grpc_subscription.rs

@@ -35,6 +35,7 @@ use tracing::debug_span;
 use crate::rpc_polling::vote_accounts_and_cluster_info_polling::{
     poll_cluster_info, poll_vote_accounts,
 };
+use solana_lite_rpc_core::structures::produced_block::ProducedBlockInner;
 use yellowstone_grpc_proto::prelude::SubscribeUpdateBlock;
 
 /// grpc version of ProducedBlock mapping
@@ -254,7 +255,7 @@ pub fn from_grpc_block_update(
         None
     };
 
-    ProducedBlock {
+    let inner = ProducedBlockInner {
         transactions: txs,
         block_height: block
             .block_height
@@ -263,12 +264,12 @@ pub fn from_grpc_block_update(
         block_time: block.block_time.map(|time| time.timestamp).unwrap() as u64,
         blockhash: block.blockhash,
         previous_blockhash: block.parent_blockhash,
-        commitment_config,
         leader_id,
         parent_slot: block.parent_slot,
         slot: block.slot,
         rewards,
-    }
+    };
+    ProducedBlock::new(inner, commitment_config)
 }
 
 pub fn create_grpc_subscription(

cluster-endpoints/src/rpc_polling/poll_blocks.rs

@@ -1,7 +1,7 @@
 use anyhow::{bail, Context};
 use solana_client::nonblocking::rpc_client::RpcClient;
 use solana_lite_rpc_core::encoding::BinaryEncoding;
-use solana_lite_rpc_core::structures::produced_block::TransactionInfo;
+use solana_lite_rpc_core::structures::produced_block::{ProducedBlockInner, TransactionInfo};
 use solana_lite_rpc_core::{
     structures::{
         produced_block::ProducedBlock,
@@ -319,7 +319,7 @@ pub fn from_ui_block(
 
     let block_time = block.block_time.unwrap_or(0) as u64;
 
-    ProducedBlock {
+    let inner = ProducedBlockInner {
         transactions: txs,
         block_height,
         leader_id,
@@ -328,9 +328,9 @@ pub fn from_ui_block(
         parent_slot,
         block_time,
         slot,
-        commitment_config,
         rewards,
-    }
+    };
+    ProducedBlock::new(inner, commitment_config)
 }
 
 #[inline]

core/src/structures/produced_block.rs

@@ -3,6 +3,9 @@ use solana_sdk::message::v0::MessageAddressTableLookup;
 use solana_sdk::pubkey::Pubkey;
 use solana_sdk::{slot_history::Slot, transaction::TransactionError};
 use solana_transaction_status::Reward;
+use std::fmt::Debug;
+use std::ops::Deref;
+use std::sync::Arc;
 
 #[derive(Debug, Clone)]
 pub struct TransactionInfo {
@@ -19,9 +22,42 @@ pub struct TransactionInfo {
     pub address_lookup_tables: Vec<MessageAddressTableLookup>,
 }
 
-// TODO try to remove Clone
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct ProducedBlock {
+    // Arc is required for channels
+    inner: Arc<ProducedBlockInner>,
+    pub commitment_config: CommitmentConfig,
+}
+
+impl ProducedBlock {
+    pub fn new(inner: ProducedBlockInner, commitment_config: CommitmentConfig) -> Self {
+        ProducedBlock {
+            inner: Arc::new(inner),
+            commitment_config,
+        }
+    }
+}
+
+/// # Example
+/// ```text
+/// ProducedBlock { slot: 254169151, commitment_config: processed, blockhash: BULfZwLswkDbHhTrHGDASUtmNAG8gk6TV2njnobjYLyd, transactions_count: 806 }
+/// ```
+impl Debug for ProducedBlock {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "ProducedBlock {{ slot: {}, commitment_config: {}, blockhash: {}, transactions_count: {} }}",
+               self.slot, self.commitment_config.commitment, self.blockhash, self.transactions.len())
+    }
+}
+
+impl Deref for ProducedBlock {
+    type Target = ProducedBlockInner;
+
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+pub struct ProducedBlockInner {
     pub transactions: Vec<TransactionInfo>,
     pub leader_id: Option<String>,
     pub blockhash: String,
@@ -29,25 +65,24 @@ pub struct ProducedBlock {
     pub slot: Slot,
     pub parent_slot: Slot,
     pub block_time: u64,
-    pub commitment_config: CommitmentConfig,
     pub previous_blockhash: String,
     pub rewards: Option<Vec<Reward>>,
 }
 
 impl ProducedBlock {
-    /// moving commitment level to finalized
-    pub fn to_finalized_block(&self) -> Self {
-        ProducedBlock {
-            commitment_config: CommitmentConfig::finalized(),
-            ..self.clone()
-        }
-    }
-
     /// moving commitment level to confirmed
     pub fn to_confirmed_block(&self) -> Self {
         ProducedBlock {
+            inner: self.inner.clone(),
             commitment_config: CommitmentConfig::confirmed(),
-            ..self.clone()
+        }
+    }
+
+    /// moving commitment level to finalized
+    pub fn to_finalized_block(&self) -> Self {
+        ProducedBlock {
+            inner: self.inner.clone(),
+            commitment_config: CommitmentConfig::finalized(),
         }
     }
 }

services/src/data_caching_service.rs

@@ -64,7 +64,7 @@ impl DataCachingService {
                     _ => TransactionConfirmationStatus::Processed,
                 };
 
-                for tx in block.transactions {
+                for tx in &block.transactions {
                     let block_info = data_cache
                         .block_information_store
                         .get_block_info(&tx.recent_blockhash);
@@ -101,7 +101,7 @@ impl DataCachingService {
                     // notify
                     data_cache
                         .tx_subs
-                        .notify(block.slot, &tx, block.commitment_config)
+                        .notify(block.slot, tx, block.commitment_config)
                         .await;
                 }
             }