blockstream store to disk

Cargo.toml

@@ -16,7 +16,9 @@ members = [
     "accounts",
     "accounts-on-demand",
     #examples
-    "examples/custom-tpu-send-transactions"
+    "examples/custom-tpu-send-transactions",
+    "examples/dump-blocks-from-grpc-to-disk",
+    "examples/query-blockstore",
 ]
 
 [workspace.package]

blockstore/Cargo.toml

@@ -23,7 +23,8 @@ solana-rpc-client-api = {workspace = true}
 native-tls = { workspace = true }
 postgres-native-tls = { workspace = true }
 anyhow = { workspace = true }
-log = {workspace = true}
+log = { workspace = true }
+tracing = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["std", "env-filter"] }
 chrono = {workspace = true}
 serde = { workspace = true }
@@ -34,10 +35,14 @@ base64 = {workspace = true}
 itertools = {workspace = true}
 rangetools = {workspace = true}
 tokio-postgres = { version = "0.7.8", features = ["with-chrono-0_4"] }
+postgres-types = { version = "0.2.6", features = ["derive", "with-serde_json-1"] }
+prometheus = { workspace = true }
 futures = {workspace = true}
 futures-util = {workspace = true}
 bytes = "1.5.0"
 rand = "0.8.5"
+lazy_static = "1.4.0"
+regex = "1.10.3"
 
 [dev-dependencies]
 tracing-subscriber = { workspace = true }

blockstore/examples/bench_postgres_simple_select.rs

@@ -3,20 +3,20 @@ use itertools::Itertools;
 /// test program to query postgres the simples possible way
 ///
 use log::info;
+use solana_lite_rpc_blockstore::block_stores::postgres::BlockstorePostgresSessionConfig;
 use solana_lite_rpc_blockstore::block_stores::postgres::PostgresSession;
-use solana_lite_rpc_blockstore::block_stores::postgres::PostgresSessionConfig;
 
 #[tokio::main(flavor = "multi_thread", worker_threads = 16)]
 pub async fn main() -> anyhow::Result<()> {
     tracing_subscriber::fmt::init();
 
-    let pg_session_config = PostgresSessionConfig::new_for_tests();
+    let pg_session_config = BlockstorePostgresSessionConfig::new_for_tests();
 
     let single_session = PostgresSession::new(pg_session_config.clone())
         .await
         .unwrap();
     // run one query
-    query_database_simple(single_session).await;
+    query_database_simple(&single_session).await;
     info!("single query test ... done");
 
     // run parallel queries
@@ -26,7 +26,7 @@ pub async fn main() -> anyhow::Result<()> {
     Ok(())
 }
 
-async fn parallel_queries(pg_session_config: PostgresSessionConfig) {
+async fn parallel_queries(pg_session_config: BlockstorePostgresSessionConfig) {
     let many_sessions = vec![
         PostgresSession::new(pg_session_config.clone())
             .await
@@ -56,15 +56,15 @@ async fn parallel_queries(pg_session_config: PostgresSessionConfig) {
 
     let futures = (0..many_sessions.len())
         .map(|si| {
-            let session = many_sessions[si].clone();
+            let session = &many_sessions[si];
             query_database_simple(session)
         })
         .collect_vec();
 
     futures_util::future::join_all(futures).await;
 }
 
-async fn query_database_simple(postgres_session: PostgresSession) {
+async fn query_database_simple(postgres_session: &PostgresSession) {
     let statement = "SELECT 1";
 
     let started = tokio::time::Instant::now();

blockstore/src/block_stores/faithful_history/faithful_block_store.rs

@@ -10,6 +10,7 @@ use solana_transaction_status::{TransactionDetails, UiTransactionEncoding};
 use std::ops::RangeInclusive;
 use std::sync::Arc;
 
+#[derive(Clone)]
 pub struct FaithfulBlockStore {
     faithful_rpc_client: Arc<RpcClient>, // to fetch legacy blocks from faithful_history
 }

blockstore/src/block_stores/multiple_strategy_block_store.rs

@@ -1,12 +1,12 @@
 use crate::block_stores::faithful_history::faithful_block_store::FaithfulBlockStore;
 use crate::block_stores::postgres::postgres_block_store_query::PostgresQueryBlockStore;
 use anyhow::{bail, Context, Result};
-use log::{debug, trace};
+
+use log::{debug, info, trace};
 use solana_lite_rpc_core::structures::produced_block::ProducedBlock;
-use solana_rpc_client::nonblocking::rpc_client::RpcClient;
+
 use solana_sdk::slot_history::Slot;
 use std::ops::{Deref, RangeInclusive};
-use std::sync::Arc;
 
 #[derive(Debug, Clone)]
 pub enum BlockSource {
@@ -18,7 +18,6 @@ pub enum BlockSource {
 
 #[derive(Debug, Clone)]
 pub struct BlockStorageData {
-    // note: commitment_config is the actual commitment level
     pub block: ProducedBlock,
     // meta data
     pub result_source: BlockSource,
@@ -32,6 +31,7 @@ impl Deref for BlockStorageData {
 }
 
 // you might need to add a read-cache instead
+// #[derive(Clone)]
 pub struct MultipleStrategyBlockStorage {
     block_storage_query: PostgresQueryBlockStore,
     // note supported ATM
@@ -40,10 +40,12 @@ pub struct MultipleStrategyBlockStorage {
 }
 
 impl MultipleStrategyBlockStorage {
-    pub fn new(
-        block_storage_query: PostgresQueryBlockStore,
-        _faithful_rpc_client: Option<Arc<RpcClient>>,
-    ) -> Self {
+    pub fn new(block_storage_query: PostgresQueryBlockStore) -> Self {
+        info!(
+            "Initializing MultipleStrategyBlockStorage {} faithful history storage",
+            "without"
+        );
+
         Self {
             block_storage_query,
             // faithful_history not used ATM
@@ -108,7 +110,7 @@ impl MultipleStrategyBlockStorage {
                     .block_storage_query
                     .query_block(slot)
                     .await
-                    .context(format!("block {} not found although it was in range", slot));
+                    .context(format!("query block {} from postgres", slot));
 
                 return lookup.map(|b| BlockStorageData {
                     block: b,