implementing history, block storage method

changes after groovies review

Cargo.toml

@@ -6,6 +6,7 @@ members = [
     "quic-forward-proxy",
     "quic-forward-proxy-integration-test",
     "cluster-endpoints",
+    "history",
     "bench"
 ]
 
@@ -55,6 +56,7 @@ rustls = { version = "=0.20.8", default-features = false }
 solana-lite-rpc-services = {path = "services", version="0.2.3"}
 solana-lite-rpc-core = {path = "core", version="0.2.3"}
 solana-lite-rpc-cluster-endpoints = {path = "cluster-endpoints", version="0.2.3"}
+solana-lite-rpc-history = {path = "history", version="0.2.3"}
 
 async-trait = "0.1.68"
 yellowstone-grpc-client = "1.9.0"

cluster-endpoints/src/rpc_polling/poll_blocks.rs

@@ -1,23 +1,12 @@
 use anyhow::Context;
 use solana_client::nonblocking::rpc_client::RpcClient;
 use solana_lite_rpc_core::{
-    structures::{
-        produced_block::{ProducedBlock, TransactionInfo},
-        slot_notification::SlotNotification,
-    },
+    structures::{produced_block::ProducedBlock, slot_notification::SlotNotification},
     AnyhowJoinHandle,
 };
 use solana_rpc_client_api::config::RpcBlockConfig;
-use solana_sdk::{
-    borsh0_10::try_from_slice_unchecked,
-    commitment_config::CommitmentConfig,
-    compute_budget::{self, ComputeBudgetInstruction},
-    slot_history::Slot,
-};
-use solana_transaction_status::{
-    option_serializer::OptionSerializer, RewardType, TransactionDetails, UiTransactionEncoding,
-    UiTransactionStatusMeta,
-};
+use solana_sdk::{commitment_config::CommitmentConfig, slot_history::Slot};
+use solana_transaction_status::{TransactionDetails, UiTransactionEncoding};
 use std::{
     sync::{atomic::AtomicU64, Arc},
     time::Duration,
@@ -42,126 +31,10 @@ pub async fn process_block(
         )
         .await;
 
-    if block.is_err() {
-        return None;
+    match block {
+        Ok(block) => Some(ProducedBlock::from_ui_block(block, slot, commitment_config)),
+        Err(_) => None,
     }
-    let block = block.unwrap();
-
-    let Some(block_height) = block.block_height else {
-        return None;
-    };
-
-    let Some(txs) = block.transactions else {
-        return None;
-    };
-
-    let blockhash = block.blockhash;
-    let parent_slot = block.parent_slot;
-
-    let txs = txs
-        .into_iter()
-        .filter_map(|tx| {
-            let Some(UiTransactionStatusMeta {
-                err,
-                compute_units_consumed,
-                ..
-            }) = tx.meta
-            else {
-                log::info!("Tx with no meta");
-                return None;
-            };
-
-            let Some(tx) = tx.transaction.decode() else {
-                log::info!("Tx could not be decoded");
-                return None;
-            };
-
-            let signature = tx.signatures[0].to_string();
-            let cu_consumed = match compute_units_consumed {
-                OptionSerializer::Some(cu_consumed) => Some(cu_consumed),
-                _ => None,
-            };
-
-            let legacy_compute_budget = tx.message.instructions().iter().find_map(|i| {
-                if i.program_id(tx.message.static_account_keys())
-                    .eq(&compute_budget::id())
-                {
-                    if let Ok(ComputeBudgetInstruction::RequestUnitsDeprecated {
-                        units,
-                        additional_fee,
-                    }) = try_from_slice_unchecked(i.data.as_slice())
-                    {
-                        return Some((units, additional_fee));
-                    }
-                }
-                None
-            });
-
-            let mut cu_requested = tx.message.instructions().iter().find_map(|i| {
-                if i.program_id(tx.message.static_account_keys())
-                    .eq(&compute_budget::id())
-                {
-                    if let Ok(ComputeBudgetInstruction::SetComputeUnitLimit(limit)) =
-                        try_from_slice_unchecked(i.data.as_slice())
-                    {
-                        return Some(limit);
-                    }
-                }
-                None
-            });
-
-            let mut prioritization_fees = tx.message.instructions().iter().find_map(|i| {
-                if i.program_id(tx.message.static_account_keys())
-                    .eq(&compute_budget::id())
-                {
-                    if let Ok(ComputeBudgetInstruction::SetComputeUnitPrice(price)) =
-                        try_from_slice_unchecked(i.data.as_slice())
-                    {
-                        return Some(price);
-                    }
-                }
-
-                None
-            });
-
-            if let Some((units, additional_fee)) = legacy_compute_budget {
-                cu_requested = Some(units);
-                if additional_fee > 0 {
-                    prioritization_fees = Some(((units * 1000) / additional_fee).into())
-                }
-            };
-
-            Some(TransactionInfo {
-                signature,
-                err,
-                cu_requested,
-                prioritization_fees,
-                cu_consumed,
-            })
-        })
-        .collect();
-
-    let leader_id = if let Some(rewards) = block.rewards {
-        rewards
-            .iter()
-            .find(|reward| Some(RewardType::Fee) == reward.reward_type)
-            .map(|leader_reward| leader_reward.pubkey.clone())
-    } else {
-        None
-    };
-
-    let block_time = block.block_time.unwrap_or(0) as u64;
-
-    Some(ProducedBlock {
-        txs,
-        block_height,
-        leader_id,
-        blockhash,
-        parent_slot,
-        block_time,
-        slot,
-        commitment_config,
-    })
 }
 
 pub fn poll_block(

core/src/commitment_utils.rs

@@ -0,0 +1,46 @@
+use solana_sdk::commitment_config::{CommitmentConfig, CommitmentLevel};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+#[repr(C)]
+pub enum Commitment {
+    Processed = 0,
+    Confirmed = 1,
+    Finalized = 2,
+}
+
+impl From<CommitmentLevel> for Commitment {
+    #[allow(deprecated)]
+    fn from(value: CommitmentLevel) -> Self {
+        match value {
+            CommitmentLevel::Finalized | CommitmentLevel::Root | CommitmentLevel::Max => {
+                Commitment::Finalized
+            }
+            CommitmentLevel::Confirmed
+            | CommitmentLevel::Single
+            | CommitmentLevel::SingleGossip => Commitment::Confirmed,
+            CommitmentLevel::Processed | CommitmentLevel::Recent => Commitment::Processed,
+        }
+    }
+}
+
+impl From<CommitmentConfig> for Commitment {
+    fn from(value: CommitmentConfig) -> Self {
+        value.commitment.into()
+    }
+}
+
+impl Commitment {
+    pub fn into_commitment_level(&self) -> CommitmentLevel {
+        match self {
+            Commitment::Confirmed => CommitmentLevel::Confirmed,
+            Commitment::Processed => CommitmentLevel::Processed,
+            Commitment::Finalized => CommitmentLevel::Finalized,
+        }
+    }
+
+    pub fn into_commiment_config(&self) -> CommitmentConfig {
+        CommitmentConfig {
+            commitment: self.into_commitment_level(),
+        }
+    }
+}

core/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod commitment_utils;
 pub mod keypair_loader;
 pub mod quic_connection;
 pub mod quic_connection_utils;

core/src/stores/data_cache.rs

@@ -28,7 +28,7 @@ pub struct SlotCache {
 /// The central data store for all data from the cluster.
 #[derive(Clone)]
 pub struct DataCache {
-    pub block_store: BlockInformationStore,
+    pub block_information_store: BlockInformationStore,
     pub txs: TxStore,
     pub tx_subs: SubscriptionStore,
     pub slot_cache: SlotCache,
@@ -39,10 +39,10 @@ pub struct DataCache {
 impl DataCache {
     pub async fn clean(&self, ttl_duration: std::time::Duration) {
         let block_info = self
-            .block_store
+            .block_information_store
             .get_latest_block_info(CommitmentConfig::finalized())
             .await;
-        self.block_store.clean().await;
+        self.block_information_store.clean().await;
         self.txs.clean(block_info.block_height);
 
         self.tx_subs.clean(ttl_duration);
@@ -55,7 +55,7 @@ impl DataCache {
         self.txs
             .is_transaction_confirmed(&sent_transaction_info.signature)
             || self
-                .block_store
+                .block_information_store
                 .get_latest_block(CommitmentConfig::processed())
                 .await
                 .block_height
@@ -64,7 +64,7 @@ impl DataCache {
 
     pub fn new_for_tests() -> Self {
         Self {
-            block_store: BlockInformationStore::new(BlockInformation {
+            block_information_store: BlockInformationStore::new(BlockInformation {
                 block_height: 0,
                 blockhash: Hash::new_unique().to_string(),
                 cleanup_slot: 1000,