saving blocks + tx to Postgres + basic multi-storage-strategy

Cargo.toml

@@ -29,13 +29,15 @@ solana-pubsub-client = "~1.16.3"
 solana-streamer = "~1.16.3"
 solana-account-decoder = "~1.16.3"
 itertools = "0.10.5"
+rangetools = "0.1.4"
 serde = { version = "1.0.160", features = ["derive"] }
 serde_json = "1.0.96"
 bincode = "1.3.3"
 bs58 = "0.4.0"
 base64 = "0.21.0"
 thiserror = "1.0.40"
 futures = "0.3.28"
+futures-util = "0.3.28"
 bytes = "1.4.0"
 anyhow = "1.0.70"
 log = "0.4.17"

cluster-endpoints/src/grpc_subscription.rs

@@ -319,10 +319,10 @@ pub fn create_grpc_subscription(
     grpc_x_token: Option<String>,
     expected_grpc_version: String,
 ) -> anyhow::Result<(EndpointStreaming, Vec<AnyhowJoinHandle>)> {
-    let (slot_sx, slot_notifier) = tokio::sync::broadcast::channel(10);
-    let (block_sx, blocks_notifier) = tokio::sync::broadcast::channel(10);
-    let (cluster_info_sx, cluster_info_notifier) = tokio::sync::broadcast::channel(10);
-    let (va_sx, vote_account_notifier) = tokio::sync::broadcast::channel(10);
+    let (slot_sx, slot_notifier) = tokio::sync::broadcast::channel(16);
+    let (block_sx, blocks_notifier) = tokio::sync::broadcast::channel(16);
+    let (cluster_info_sx, cluster_info_notifier) = tokio::sync::broadcast::channel(16);
+    let (va_sx, vote_account_notifier) = tokio::sync::broadcast::channel(16);
 
     let mut slots = HashMap::new();
     slots.insert(

cluster-endpoints/src/json_rpc_subscription.rs

@@ -12,17 +12,18 @@
 
 pub fn create_json_rpc_polling_subscription(
     rpc_client: Arc<RpcClient>,
+    num_parallel_tasks: usize,
 ) -> anyhow::Result<(EndpointStreaming, Vec<AnyhowJoinHandle>)> {
-    let (slot_sx, slot_notifier) = tokio::sync::broadcast::channel(10);
-    let (block_sx, blocks_notifier) = tokio::sync::broadcast::channel(10);
-    let (cluster_info_sx, cluster_info_notifier) = tokio::sync::broadcast::channel(10);
-    let (va_sx, vote_account_notifier) = tokio::sync::broadcast::channel(10);
+    let (slot_sx, slot_notifier) = tokio::sync::broadcast::channel(16);
+    let (block_sx, blocks_notifier) = tokio::sync::broadcast::channel(16);
+    let (cluster_info_sx, cluster_info_notifier) = tokio::sync::broadcast::channel(16);
+    let (va_sx, vote_account_notifier) = tokio::sync::broadcast::channel(16);
 
     let mut endpoint_tasks =
         poll_slots(rpc_client.clone(), CommitmentConfig::processed(), slot_sx)?;
 
     let mut block_polling_tasks =
-        poll_block(rpc_client.clone(), block_sx, slot_notifier.resubscribe());
+        poll_block(rpc_client.clone(), block_sx, slot_notifier.resubscribe(), num_parallel_tasks);
     endpoint_tasks.append(&mut block_polling_tasks);
 
     let cluster_info_polling =

cluster-endpoints/src/rpc_polling/poll_blocks.rs

@@ -16,6 +16,8 @@ use solana_transaction_status::{TransactionDetails, UiTransactionEncoding};
 use std::{sync::Arc, time::Duration};
 use tokio::sync::broadcast::{Receiver, Sender};
 
+pub const NUM_PARALLEL_TASKS_DEFAULT: usize = 16;
+
 pub async fn process_block(
     rpc_client: &RpcClient,
     slot: Slot,
@@ -42,6 +44,7 @@ pub fn poll_block(
     rpc_client: Arc<RpcClient>,
     block_notification_sender: Sender<ProducedBlock>,
     slot_notification: Receiver<SlotNotification>,
+    num_parallel_tasks: usize,
 ) -> Vec<AnyhowJoinHandle> {
     let mut tasks: Vec<AnyhowJoinHandle> = vec![];
 
@@ -50,7 +53,7 @@ pub fn poll_block(
     let (block_schedule_queue_sx, block_schedule_queue_rx) =
         async_channel::unbounded::<(Slot, CommitmentConfig)>();
 
-    for _i in 0..16 {
+    for _i in 0..num_parallel_tasks {
         let block_notification_sender = block_notification_sender.clone();
         let rpc_client = rpc_client.clone();
         let block_schedule_queue_rx = block_schedule_queue_rx.clone();

cluster-endpoints/src/rpc_polling/poll_slots.rs

@@ -78,7 +78,7 @@ pub fn poll_slots(
                 }
                 Ok(None) => log::error!("got nothing from slot update notifier"),
                 Err(err) => {
-                    log::warn!("failed to receive slot update: {err}");
+                    log::debug!("timeout on receive slot update: {err}");
                     // force update the slot
                     // estimated slot should not go ahead more than 32 slots
                     // this is because it may be a slot block

core/src/iterutils.rs

@@ -0,0 +1,17 @@
+pub enum Uniqueness {
+    ExactlyOne,
+    Multiple(usize),
+    Empty,
+}
+
+impl Uniqueness {
+    pub fn inspect_len(len: usize) -> Uniqueness {
+        if len == 1 {
+            Uniqueness::ExactlyOne
+        } else if len == 0 {
+            Uniqueness::Empty
+        } else {
+            Uniqueness::Multiple(len)
+        }
+    }
+}

core/src/lib.rs

@@ -1,5 +1,6 @@
 pub mod commitment_utils;
 pub mod encoding;
+pub mod iterutils;
 pub mod keypair_loader;
 pub mod quic_connection;
 pub mod quic_connection_utils;

core/src/stores/block_information_store.rs

@@ -1,11 +1,8 @@
 use dashmap::DashMap;
 use log::info;
 
-use solana_sdk::{
-    clock::MAX_RECENT_BLOCKHASHES,
-    commitment_config::{CommitmentConfig, CommitmentLevel},
-    slot_history::Slot,
-};
+use solana_sdk::commitment_config::{CommitmentConfig, CommitmentLevel};
+use solana_sdk::{clock::MAX_RECENT_BLOCKHASHES, slot_history::Slot};
 use std::sync::Arc;
 use tokio::sync::RwLock;
 

core/src/structures/epoch.rs

@@ -4,6 +4,7 @@ use solana_rpc_client::nonblocking::rpc_client::RpcClient;
 use solana_sdk::epoch_info::EpochInfo;
 use solana_sdk::slot_history::Slot;
 use solana_sdk::sysvar::epoch_schedule::EpochSchedule;
+use std::fmt::Display;
 use std::sync::Arc;
 
 #[derive(Debug, Copy, Clone, PartialOrd, PartialEq, Eq, Ord)]
@@ -14,8 +15,11 @@ pub struct Epoch {
     pub absolute_slot: Slot,
 }
 
+#[derive(Debug, Copy, Clone, PartialOrd, PartialEq, Eq, Ord, Hash)]
+pub struct EpochRef(u64);
+
 impl Epoch {
-    pub fn into_epoch_info(&self, block_height: u64, transaction_count: Option<u64>) -> EpochInfo {
+    pub fn as_epoch_info(&self, block_height: u64, transaction_count: Option<u64>) -> EpochInfo {
         EpochInfo {
             epoch: self.epoch,
             slot_index: self.slot_index,
@@ -27,6 +31,32 @@ impl Epoch {
     }
 }
 
+impl Display for EpochRef {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.get_epoch())
+    }
+}
+
+impl From<Epoch> for EpochRef {
+    fn from(epoch: Epoch) -> Self {
+        Self(epoch.epoch)
+    }
+}
+
+impl EpochRef {
+    pub fn new(epoch: u64) -> Self {
+        Self(epoch)
+    }
+
+    pub fn get_epoch(&self) -> u64 {
+        self.0
+    }
+
+    pub fn get_next_epoch(&self) -> Self {
+        Self(self.0 + 1)
+    }
+}
+
 #[derive(Clone)]
 pub struct EpochCache {
     epoch_schedule: Arc<EpochSchedule>,

core/src/structures/produced_block.rs

@@ -1,6 +1,6 @@
+use solana_sdk::commitment_config::CommitmentConfig;
 use solana_sdk::{
     borsh0_10::try_from_slice_unchecked,
-    commitment_config::CommitmentConfig,
     compute_budget::{self, ComputeBudgetInstruction},
     slot_history::Slot,
     transaction::TransactionError,
@@ -23,7 +23,8 @@
     pub message: String,
 }
 
-#[derive(Default, Debug, Clone)]
+// TODO try to remove Clone
+#[derive(Debug, Clone)]
 pub struct ProducedBlock {
     pub transactions: Vec<TransactionInfo>,
     pub leader_id: Option<String>,
@@ -122,7 +123,7 @@
                 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())
+                        prioritization_fees = Some(calc_prioritization_fees(units, additional_fee))
                     }
                 };
 
@@ -162,7 +163,23 @@
             block_time,
             slot,
             commitment_config,
             rewards,
         }
     }
+
+
+}
+
+fn calc_prioritization_fees(units: u32, additional_fee: u32) -> u64 {
+    (units as u64 * 1000) /  additional_fee as u64
+}
+
+#[test]
+fn overflow_u32() {
+    // value high enough to overflow u32 if multiplied by 1000
+    let units: u32 = 4_000_000_000;
+    let additional_fee: u32 = 100;
+    let prioritization_fees: u64 = calc_prioritization_fees(units, additional_fee);
+
+    assert_eq!(40_000_000_000, prioritization_fees);
 }

core/src/traits/mod.rs

@@ -1,3 +1,2 @@
-pub mod block_storage_interface;
 pub mod leaders_fetcher_interface;
 pub mod subscription_sink;

history/Cargo.toml

@@ -14,17 +14,30 @@ solana-rpc-client = { workspace = true }
 
 dashmap = {workspace = true}
 async-trait = { workspace = true }
-tokio = "1.*"
+tokio = { version = "1.28.2", features = ["full", "fs"]}
+tokio-util = "0.7"
 
 solana-lite-rpc-core = {workspace = true}
+solana-lite-rpc-cluster-endpoints = {workspace = true}
 solana-rpc-client-api = {workspace = true}
 native-tls = { workspace = true }
 postgres-native-tls = { workspace = true }
 anyhow = { workspace = true }
 log = {workspace = true}
+tracing-subscriber = { workspace = true, features = ["std", "env-filter"] }
 chrono = {workspace = true}
+serde = { workspace = true }
+serde_json = { workspace = true }
+jsonrpsee = { workspace = true }
 bincode = {workspace = true}
 base64 = {workspace = true}
 itertools = {workspace = true}
+rangetools = {workspace = true}
 tokio-postgres = { version = "0.7.8", features = ["with-chrono-0_4"] }
-serde = { workspace = true }
+futures = {workspace = true}
+futures-util = {workspace = true}
+bytes = "1.5.0"
+rand = "0.8.5"
+
+[dev-dependencies]
+tracing-subscriber = { workspace = true }